mirror of
https://github.com/hackdiaz-dev/free-turn-proxy.git
synced 2026-10-12 09:41:53 +02:00
fix(session): рецикл после пробуждения только при тишине в канале
гэп-детектор сносил живые аллокации: сон сам по себе их не убивает, а keepalive туннеля идёт через тот же relay. пинок теперь проверяет rx в окне и схлопывает пачку сигналов в один рецикл.
This commit is contained in:
samosvalishe
@@ -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()
|
||||
|
||||
@@ -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):
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user