fix: soft reconnect on network handover without tunnel drop

2026-10-12 09:41:53 +02:00 · 2026-08-17 23:29:19 +03:00
parent 5fdd8c3112
commit 0bef2e8ab5
4 changed files with 54 additions and 39 deletions
@@ -37,7 +37,6 @@ func DTLSLoop(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr
 				select {
 				case <-ctx.Done():
 					return
-				case <-deps.woke():
 				case <-time.After(time.Duration(10+randx.Intn(20)) * time.Second):
 				}
 			}
@@ -247,9 +246,6 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
 		case <-stream.PermDead:
 			deps.log().Warnf("[STREAM %d] TURN channel-bind умер - рецикл allocation", streamID)
 			turncancel()
-		case <-deps.woke():
-			deps.log().Warnf("[STREAM %d] Пробуждение устройства - рецикл allocation", streamID)
-			turncancel()
 		}
 		// conn2 молчит, пока приложение не шлёт: без дедлайна его читатель досидел бы
 		// до первого пакета, а рецикл на простое висел бы вечно (тоннель без трафика)
@@ -12,7 +12,6 @@ import (
 	"github.com/samosvalishe/free-turn-proxy/internal/logx"
 	"github.com/samosvalishe/free-turn-proxy/internal/stats"
 	"github.com/samosvalishe/free-turn-proxy/internal/transport/dtlsdial"
-	"github.com/samosvalishe/free-turn-proxy/internal/wake"
 )

 // AuthHandler определяет интерфейс взаимодействия с провайдером при ошибках авторизации.
@@ -48,7 +47,6 @@ type Deps struct {
 	ActiveLocalPeer  *atomic.Value
 	ConnectedStreams *atomic.Int32
 	OnTURNServer     func(ip net.IP)
-	Wake             *wake.Notifier
 	fatalCh          chan error
 }

@@ -59,13 +57,11 @@ func (d *Deps) log() logx.Logger {
 	return d.Log
 }

-func (d *Deps) woke() <-chan struct{} { return d.Wake.Chan() }
-
 // Run запускает прием входящего UDP-трафика и распределяет его по пулу пар DTLSLoop/TURNLoop.
-func Run(ctx context.Context, dtlsDialer *dtlsdial.Dialer, auth AuthHandler, logger logx.Logger, connectedStreams *atomic.Int32, onTURNServer func(net.IP), waker *wake.Notifier, params *Params, peer *net.UDPAddr, listenConn net.PacketConn, numStreams int) error {
+func Run(ctx context.Context, dtlsDialer *dtlsdial.Dialer, auth AuthHandler, logger logx.Logger, connectedStreams *atomic.Int32, onTURNServer func(net.IP), params *Params, peer *net.UDPAddr, listenConn net.PacketConn, numStreams int) error {
 	context.AfterFunc(ctx, func() {
-		if closeErr := listenConn.Close(); closeErr != nil {
-			logger.Errorf("udprelay: close local connection: %s", closeErr)
+		if err := listenConn.SetReadDeadline(time.Now()); err != nil {
+			logger.Errorf("udprelay: set listen deadline: %s", err)
 		}
 	})

@@ -82,7 +78,6 @@ func Run(ctx context.Context, dtlsDialer *dtlsdial.Dialer, auth AuthHandler, log
 		ActiveLocalPeer:  &activeLocalPeer,
 		ConnectedStreams: connectedStreams,
 		OnTURNServer:     onTURNServer,
-		Wake:             waker,
 		fatalCh:          fatalCh,
 	}

@@ -137,6 +132,11 @@ func Run(ctx context.Context, dtlsDialer *dtlsdial.Dialer, auth AuthHandler, log
 	}()

 	wg.Wait()
+	// Сбрасываем дедлайн, чтобы listenConn можно было переиспользовать
+	// в следующей итерации runRelayLoop (актуально для LocalPipe).
+	if err := listenConn.SetReadDeadline(time.Time{}); err != nil {
+		logger.Errorf("udprelay: clear listen deadline: %s", err)
+	}
 	runCancel()
 	<-watcherDone
 	if p := fatalErr.Load(); p != nil {
@@ -19,7 +19,6 @@ import (
 	"github.com/samosvalishe/free-turn-proxy/internal/routemgr"
 	"github.com/samosvalishe/free-turn-proxy/internal/stats"
 	"github.com/samosvalishe/free-turn-proxy/internal/transport/dtlsdial"
-	"github.com/samosvalishe/free-turn-proxy/internal/wake"
 )

 // Phase - текущая стадия подключения сессии.
@@ -40,14 +39,9 @@ var (
 )

 const (
-	defaultStatusInterval       = 2 * time.Second
+	defaultStatusInterval       = 500 * time.Millisecond
 	defaultUDPHandshakeTimeout  = 20 * time.Second
 	defaultHandshakeConcurrency = 3
-
-	// Порог вдвое больше тика: сон короче порога всё равно пропускаем, а тик почаще
-	// стоил бы пробуждений процесса на всё время сессии.
-	wakeTick      = 30 * time.Second
-	wakeThreshold = 60 * time.Second
 )

 type Options struct {
@@ -110,10 +104,10 @@ type Session struct {
 	total   int
 	traffic *traffic

-	connected atomic.Int32
-	status    atomic.Pointer[statusInfo]
-	started   atomic.Bool
-	wake      *wake.Notifier
+	connected   atomic.Int32
+	status      atomic.Pointer[statusInfo]
+	started     atomic.Bool
+	reconnectCh chan struct{}
 }

 func New(cfg *config.Client, deps Deps) (*Session, error) {
@@ -127,11 +121,11 @@ func New(cfg *config.Client, deps Deps) (*Session, error) {
 	total := cfg.TURN.N * max(len(cfg.VK.Links), 1)

 	s := &Session{
-		cfg:   cfg,
-		deps:  deps,
-		opts:  deps.Options.withDefaults(),
-		total: total,
-		wake:  wake.New(),
+		cfg:         cfg,
+		deps:        deps,
+		opts:        deps.Options.withDefaults(),
+		total:       total,
+		reconnectCh: make(chan struct{}, 1),
 	}
 	if s.opts.Traffic {
 		s.traffic = newTraffic()
@@ -181,18 +175,12 @@ func (s *Session) Run(ctx context.Context) (err error) {
 	defer cancel()

 	var bg sync.WaitGroup
-	bg.Go(func() {
-		s.wake.Watch(runCtx, wakeTick, wakeThreshold, func(gap time.Duration) {
-			log.Warnf("device slept for %s - recycling TURN allocations", gap.Truncate(time.Second))
-		})
-	})
-
 	var watchdogErr error
 	bg.Go(func() {
 		watchdogErr = s.watch(runCtx, cancel)
 	})

-	relayErr := s.relay(runCtx, prov, peer)
+	relayErr := s.runRelayLoop(runCtx, prov, peer)
 	stopped := runCtx.Err() != nil
 	cancel()
 	bg.Wait()
@@ -203,8 +191,37 @@ func (s *Session) Run(ctx context.Context) (err error) {
 	return watchdogErr
 }

-// Wake инициирует немедленное переподключение TURN-аллокаций после пробуждения.
-func (s *Session) Wake() { s.wake.Fire() }
+func (s *Session) runRelayLoop(ctx context.Context, prov provider.Provider, peer *net.UDPAddr) error {
+	for {
+		attemptCtx, attemptCancel := context.WithCancel(ctx)
+		done := make(chan error, 1)
+		go func() { done <- s.relay(attemptCtx, prov, peer) }()
+
+		var err error
+		select {
+		case err = <-done:
+		case <-s.reconnectCh:
+			attemptCancel()
+			err = <-done
+		}
+		attemptCancel()
+
+		if ctx.Err() != nil {
+			return err
+		}
+		if err != nil && !errors.Is(err, context.Canceled) {
+			return err
+		}
+	}
+}
+
+// Wake форсирует чистый перезапуск relay (например, при смене сети), не трогая сам туннель.
+func (s *Session) Wake() {
+	select {
+	case s.reconnectCh <- struct{}{}:
+	default:
+	}
+}

 // Snapshot возвращает текущий снимок состояния сессии.
 func (s *Session) Snapshot() Snapshot {
@@ -305,6 +322,9 @@ func (s *Session) relay(ctx context.Context, prov provider.Provider, peer *net.U
 	if err != nil {
 		return err
 	}
+	if s.deps.LocalPipe == nil {
+		defer local.Close()
+	}

 	dialer := &dtlsdial.Dialer{
 		HandshakeTimeout: s.opts.UDPHandshakeTimeout,
@@ -321,7 +341,7 @@ func (s *Session) relay(ctx context.Context, prov provider.Provider, peer *net.U
 		ClientID:     s.cfg.ClientID,
 		TrafficStats: s.trafficStats(),
 	}
-	return udprelay.Run(ctx, dialer, prov, log, &s.connected, routeCallback, s.wake, params, peer, local, s.total)
+	return udprelay.Run(ctx, dialer, prov, log, &s.connected, routeCallback, params, peer, local, s.total)
 }

 func (s *Session) localConn(ctx context.Context) (net.PacketConn, error) {
@@ -278,7 +278,6 @@ func startLocked(configJSON string, tunFD int, withTunnel bool) error {
 		}
 	}

-	ClearLogs()
 	ctx, cancel := context.WithCancel(context.Background())
 	l := &live{
 		sess:   sess,