mirror of
https://github.com/hackdiaz-dev/free-turn-proxy.git
synced 2026-10-12 09:41:53 +02:00
fix(udprelay): не зависать на фатальной ошибке провайдера
Дедлайн, будивший runListener, ставился по родительскому ctx, а фатальная ошибка отменяет только runCtx: на молчащем LocalPipe wg.Wait() не возвращался никогда.
This commit is contained in:
@@ -60,12 +60,6 @@ func (d *Deps) log() logx.Logger {
|
||||
|
||||
// Run запускает прием входящего UDP-трафика и распределяет его по пулу пар DTLSLoop/TURNLoop.
|
||||
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 err := listenConn.SetReadDeadline(time.Now()); err != nil {
|
||||
logger.Errorf("udprelay: set listen deadline: %s", err)
|
||||
}
|
||||
})
|
||||
|
||||
if numStreams <= 0 {
|
||||
numStreams = 1
|
||||
}
|
||||
@@ -85,6 +79,15 @@ func Run(ctx context.Context, dtlsDialer *dtlsdial.Dialer, auth AuthHandler, log
|
||||
runCtx, runCancel := context.WithCancel(ctx)
|
||||
defer runCancel()
|
||||
|
||||
deadlineSet := make(chan struct{})
|
||||
go func() {
|
||||
defer close(deadlineSet)
|
||||
<-runCtx.Done()
|
||||
if err := listenConn.SetReadDeadline(time.Now()); err != nil {
|
||||
logger.Errorf("udprelay: set listen deadline: %s", err)
|
||||
}
|
||||
}()
|
||||
|
||||
inboundChan := make(chan *Packet, inboundQueueCap)
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Go(func() {
|
||||
@@ -133,12 +136,11 @@ func Run(ctx context.Context, dtlsDialer *dtlsdial.Dialer, auth AuthHandler, log
|
||||
}()
|
||||
|
||||
wg.Wait()
|
||||
// Сбрасываем дедлайн, чтобы listenConn можно было переиспользовать
|
||||
// в следующей итерации runRelayLoop (актуально для LocalPipe).
|
||||
runCancel()
|
||||
<-deadlineSet
|
||||
if err := listenConn.SetReadDeadline(time.Time{}); err != nil {
|
||||
logger.Errorf("udprelay: clear listen deadline: %s", err)
|
||||
}
|
||||
runCancel()
|
||||
<-watcherDone
|
||||
if p := fatalErr.Load(); p != nil {
|
||||
return *p
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
package udprelay
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/samosvalishe/free-turn-proxy/internal/logx"
|
||||
"github.com/samosvalishe/free-turn-proxy/internal/netconn"
|
||||
"github.com/samosvalishe/free-turn-proxy/internal/provider"
|
||||
"github.com/samosvalishe/free-turn-proxy/internal/transport/dtlsdial"
|
||||
)
|
||||
|
||||
type stubAuth struct{}
|
||||
|
||||
func (stubAuth) IsAuthError(error) bool { return false }
|
||||
func (stubAuth) HandleAuthError(int) bool { return false }
|
||||
func (stubAuth) ResetErrors(int) {}
|
||||
func (stubAuth) DropCredentials(int) {}
|
||||
func (stubAuth) BackoffUntilUnix() int64 { return 0 }
|
||||
|
||||
type deadlineRecorder struct {
|
||||
net.PacketConn
|
||||
mu sync.Mutex
|
||||
last time.Time
|
||||
set bool
|
||||
}
|
||||
|
||||
func (d *deadlineRecorder) SetReadDeadline(t time.Time) error {
|
||||
d.mu.Lock()
|
||||
d.last, d.set = t, true
|
||||
d.mu.Unlock()
|
||||
return d.PacketConn.SetReadDeadline(t)
|
||||
}
|
||||
|
||||
func (d *deadlineRecorder) lastDeadline() (time.Time, bool) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
return d.last, d.set
|
||||
}
|
||||
|
||||
func runFatalDeps(t *testing.T) (*dtlsdial.Dialer, *Params, *net.UDPAddr, *deadlineRecorder) {
|
||||
t.Helper()
|
||||
|
||||
pipe, peerSide := netconn.PacketPipe(1500, 4)
|
||||
t.Cleanup(func() { _ = pipe.Close(); _ = peerSide.Close() })
|
||||
|
||||
params := &Params{
|
||||
GetCreds: func(context.Context, int) (string, string, []string, error) {
|
||||
return "", "", nil, provider.ErrFatalNoStreams
|
||||
},
|
||||
}
|
||||
return &dtlsdial.Dialer{HandshakeTimeout: 100 * time.Millisecond},
|
||||
params,
|
||||
&net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 9},
|
||||
&deadlineRecorder{PacketConn: pipe}
|
||||
}
|
||||
|
||||
// Фатальная ошибка провайдера отменяет только runCtx, поэтому будить runListener обязан
|
||||
// сам Run: на молчащем LocalPipe тот сидит в ReadFrom, и без этого wg.Wait висит вечно.
|
||||
func TestRunReturnsOnFatalProviderError(t *testing.T) {
|
||||
t.Parallel()
|
||||
dialer, params, peer, local := runFatalDeps(t)
|
||||
|
||||
var connected atomic.Int32
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- Run(context.Background(), dialer, stubAuth{}, logx.Nop(), &connected, nil, params, peer, local, 1)
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
if !errors.Is(err, ErrFatal) {
|
||||
t.Fatalf("err = %v, want ErrFatal", err)
|
||||
}
|
||||
case <-time.After(30 * time.Second):
|
||||
t.Fatal("Run hung on fatal provider error")
|
||||
}
|
||||
|
||||
// Тот же LocalPipe достаётся следующей попытке: просроченный дедлайн порвал бы её чтение.
|
||||
last, set := local.lastDeadline()
|
||||
if !set {
|
||||
t.Fatal("read deadline was never touched")
|
||||
}
|
||||
if !last.IsZero() {
|
||||
t.Fatalf("read deadline left at %v, want cleared", last)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user