From 9c2cc93a1caa2ecbea87be8501dd13e6fb358071 Mon Sep 17 00:00:00 2001 From: samosvalishe Date: Tue, 18 Aug 2026 15:02:33 +0300 Subject: [PATCH] =?UTF-8?q?fix(session):=20=D1=80=D0=B5=D1=86=D0=B8=D0=BA?= =?UTF-8?q?=D0=BB=20=D0=BF=D0=BE=D1=81=D0=BB=D0=B5=20=D0=BF=D1=80=D0=BE?= =?UTF-8?q?=D0=B1=D1=83=D0=B6=D0=B4=D0=B5=D0=BD=D0=B8=D1=8F=20=D1=82=D0=BE?= =?UTF-8?q?=D0=BB=D1=8C=D0=BA=D0=BE=20=D0=BF=D1=80=D0=B8=20=D1=82=D0=B8?= =?UTF-8?q?=D1=88=D0=B8=D0=BD=D0=B5=20=D0=B2=20=D0=BA=D0=B0=D0=BD=D0=B0?= =?UTF-8?q?=D0=BB=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit гэп-детектор сносил живые аллокации: сон сам по себе их не убивает, а keepalive туннеля идёт через тот же relay. пинок теперь проверяет rx в окне и схлопывает пачку сигналов в один рецикл. --- internal/session/session.go | 70 ++++++++++++++++++++++++-- internal/session/wake_test.go | 93 +++++++++++++++++++++++++++++++++++ 2 files changed, 160 insertions(+), 3 deletions(-) create mode 100644 internal/session/wake_test.go diff --git a/internal/session/session.go b/internal/session/session.go index 120941a..0a68605 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -48,6 +48,10 @@ const ( // стоил бы пробуждений процесса на всё время сессии. wakeTick = 30 * time.Second wakeThreshold = 60 * time.Second + + // Окно проверки живости после пробуждения: рецикл - только если за него не пришло + // ни байта. Больше периода keepalive туннеля (25 c у WireGuard по умолчанию). + wakeProbeWindow = 30 * time.Second ) type Options struct { @@ -56,6 +60,8 @@ type Options struct { StatusInterval time.Duration UDPHandshakeTimeout time.Duration HandshakeConcurrency int + // WakeProbeWindow - окно проверки живости канала после пробуждения. + WakeProbeWindow time.Duration Traffic bool } @@ -70,6 +76,9 @@ func (o Options) withDefaults() Options { if o.HandshakeConcurrency <= 0 { o.HandshakeConcurrency = defaultHandshakeConcurrency } + if o.WakeProbeWindow <= 0 { + o.WakeProbeWindow = wakeProbeWindow + } return o } @@ -114,6 +123,7 @@ type Session struct { status atomic.Pointer[statusInfo] started atomic.Bool reconnectCh chan struct{} + wakeCh chan struct{} } func New(cfg *config.Client, deps Deps) (*Session, error) { @@ -132,6 +142,7 @@ func New(cfg *config.Client, deps Deps) (*Session, error) { opts: deps.Options.withDefaults(), total: total, reconnectCh: make(chan struct{}, 1), + wakeCh: make(chan struct{}, 1), } if s.opts.Traffic { s.traffic = newTraffic() @@ -183,10 +194,11 @@ func (s *Session) Run(ctx context.Context) (err error) { var bg sync.WaitGroup bg.Go(func() { wake.New().Watch(runCtx, wakeTick, wakeThreshold, func(gap time.Duration) { - log.Warnf("device slept for %s - recycling TURN allocations", gap.Truncate(time.Second)) + log.Warnf("device slept for %s - checking TURN allocations", gap.Truncate(time.Second)) s.Wake() }) }) + bg.Go(func() { s.watchWake(runCtx) }) var watchdogErr error bg.Go(func() { @@ -228,14 +240,66 @@ func (s *Session) runRelayLoop(ctx context.Context, prov provider.Provider, peer } } -// Wake форсирует чистый перезапуск relay (например, при смене сети), не трогая сам туннель. +// Wake сообщает о подозрении на сон или смену сети. Рецикл не мгновенный: сначала +// watchWake проверяет, молчит ли канал - пересоздание живых аллокаций стоит похода в VK +// за реквизитами и решения капчи. func (s *Session) Wake() { select { - case s.reconnectCh <- struct{}{}: + case s.wakeCh <- struct{}{}: default: } } +// Рецикл только по факту тишины: сон сам по себе аллокацию не убивает (lifetime 600 c), +// а keepalive туннеля идёт через тот же relay и растит Rx. Заодно схлопывает каскад - +// гэп-детектор ядра и SCREEN_ON с платформы приходят почти одновременно. +func (s *Session) watchWake(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case <-s.wakeCh: + } + if s.wakeNeedsRecycle(ctx) { + select { + case s.reconnectCh <- struct{}{}: + default: + } + } + } +} + +func (s *Session) wakeNeedsRecycle(ctx context.Context) bool { + log := s.deps.Logger + // Подключение ещё идёт: рецикл отменил бы перебор реквизитов и решение капчи. + if s.connected.Load() == 0 { + log.Warnf("wake: сессия ещё поднимается - рецикл пропущен") + return false + } + // Без счётчиков трафика подтвердить живость нечем. + if s.traffic == nil { + return true + } + _, before := s.traffic.stats.Counters() + select { + case <-ctx.Done(): + return false + case <-time.After(s.opts.WakeProbeWindow): + } + // Пока ждали, могло прилететь ещё одно пробуждение - оно про тот же сон. + select { + case <-s.wakeCh: + default: + } + _, after := s.traffic.stats.Counters() + if after > before { + log.Warnf("wake: канал жив (+%d B за %s) - рецикл не нужен", after-before, s.opts.WakeProbeWindow) + return false + } + log.Warnf("wake: тишина %s - рецикл TURN-аллокаций", s.opts.WakeProbeWindow) + return true +} + // Snapshot возвращает текущий снимок состояния сессии. func (s *Session) Snapshot() Snapshot { st := s.status.Load() diff --git a/internal/session/wake_test.go b/internal/session/wake_test.go new file mode 100644 index 0000000..c62397e --- /dev/null +++ b/internal/session/wake_test.go @@ -0,0 +1,93 @@ +package session + +import ( + "context" + "testing" + "time" + + "github.com/samosvalishe/free-turn-proxy/internal/config" +) + +func newWakeSession(t *testing.T) *Session { + t.Helper() + s, err := New(&config.Client{}, Deps{ + Options: Options{Traffic: true, WakeProbeWindow: 50 * time.Millisecond}, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + return s +} + +// Пробуждение при живом канале не должно стоить рецикла: аллокация переживает сон, а её +// пересоздание тянет за собой поход в VK и капчу. +func TestWakeSkipsRecycleWhenTrafficFlows(t *testing.T) { + t.Parallel() + s := newWakeSession(t) + s.connected.Store(1) + + done := make(chan bool, 1) + go func() { done <- s.wakeNeedsRecycle(context.Background()) }() + + // Имитируем keepalive туннеля, идущий через relay. + time.Sleep(10 * time.Millisecond) + s.traffic.stats.AddRx(64) + + select { + case need := <-done: + if need { + t.Fatal("recycle requested while traffic was flowing") + } + case <-time.After(2 * time.Second): + t.Fatal("wakeNeedsRecycle did not finish") + } +} + +func TestWakeRecyclesOnSilence(t *testing.T) { + t.Parallel() + s := newWakeSession(t) + s.connected.Store(1) + + if !s.wakeNeedsRecycle(context.Background()) { + t.Fatal("silent channel must be recycled") + } +} + +// Пробуждение во время подъёма сессии отменило бы перебор реквизитов и решение капчи. +func TestWakeSkipsRecycleWhileConnecting(t *testing.T) { + t.Parallel() + s := newWakeSession(t) + + start := time.Now() + if s.wakeNeedsRecycle(context.Background()) { + t.Fatal("recycle requested while session was still connecting") + } + if elapsed := time.Since(start); elapsed > 40*time.Millisecond { + t.Fatalf("probe window was waited unnecessarily (%s)", elapsed) + } +} + +// Гэп-детектор ядра и пинок с платформы приходят почти одновременно: второй сигнал +// про тот же сон не должен давать второй рецикл. +func TestWakeCollapsesBurst(t *testing.T) { + t.Parallel() + s := newWakeSession(t) + s.connected.Store(1) + + go s.watchWake(t.Context()) + + s.Wake() + s.Wake() + + select { + case <-s.reconnectCh: + case <-time.After(2 * time.Second): + t.Fatal("no recycle after wake on a silent channel") + } + + select { + case <-s.reconnectCh: + t.Fatal("burst produced a second recycle") + case <-time.After(200 * time.Millisecond): + } +}