From 48631616f143bc07f93d4a6cf6d5d2647022129e Mon Sep 17 00:00:00 2001 From: samosvalishe Date: Tue, 25 Aug 2026 01:56:14 +0300 Subject: [PATCH] =?UTF-8?q?fix(udprelay):=20=D0=BD=D0=B5=20=D0=B7=D0=B0?= =?UTF-8?q?=D0=B2=D0=B8=D1=81=D0=B0=D1=82=D1=8C=20=D0=BD=D0=B0=20=D1=84?= =?UTF-8?q?=D0=B0=D1=82=D0=B0=D0=BB=D1=8C=D0=BD=D0=BE=D0=B9=20=D0=BE=D1=88?= =?UTF-8?q?=D0=B8=D0=B1=D0=BA=D0=B5=20=D0=BF=D1=80=D0=BE=D0=B2=D0=B0=D0=B9?= =?UTF-8?q?=D0=B4=D0=B5=D1=80=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Дедлайн, будивший runListener, ставился по родительскому ctx, а фатальная ошибка отменяет только runCtx: на молчащем LocalPipe wg.Wait() не возвращался никогда. --- internal/proxy/udprelay/run.go | 20 ++++--- internal/proxy/udprelay/run_test.go | 92 +++++++++++++++++++++++++++++ 2 files changed, 103 insertions(+), 9 deletions(-) create mode 100644 internal/proxy/udprelay/run_test.go diff --git a/internal/proxy/udprelay/run.go b/internal/proxy/udprelay/run.go index 6df612c..52e3372 100644 --- a/internal/proxy/udprelay/run.go +++ b/internal/proxy/udprelay/run.go @@ -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 diff --git a/internal/proxy/udprelay/run_test.go b/internal/proxy/udprelay/run_test.go new file mode 100644 index 0000000..1d650bf --- /dev/null +++ b/internal/proxy/udprelay/run_test.go @@ -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) + } +}