diff --git a/internal/transport/turndial/turndial.go b/internal/transport/turndial/turndial.go index b8327c6..29d32c3 100644 --- a/internal/transport/turndial/turndial.go +++ b/internal/transport/turndial/turndial.go @@ -86,7 +86,7 @@ func Open(ctx context.Context, cfg Config, peer *net.UDPAddr, user, pass, rawAdd closeConn func() error ) if cfg.TransportUDP { - raw, derr := (&net.Dialer{Control: netctl.Apply}).Dial("udp", turnServerUDPAddr.String()) + raw, derr := (&net.Dialer{Control: netctl.Apply}).DialContext(ctx, "udp", turnServerUDPAddr.String()) if derr != nil { return nil, fmt.Errorf("dial TURN (udp): %w", derr) } @@ -150,7 +150,18 @@ func Open(ctx context.Context, cfg Config, peer *net.UDPAddr, user, pass, rawAdd } return nil, fmt.Errorf("TURN listen: %w", err) } + // Allocate не принимает ctx: отмену доставляет закрытие клиента и транспорта. + stopCancel := context.AfterFunc(ctx, func() { + client.Close() + _ = closeConn() + }) relay, err := client.Allocate() + if !stopCancel() { + if err == nil { + _ = relay.Close() + } + return nil, fmt.Errorf("TURN allocate: %w", ctx.Err()) + } if err != nil { client.Close() if cerr := closeConn(); cerr != nil { diff --git a/internal/transport/turndial/turndial_test.go b/internal/transport/turndial/turndial_test.go index facdbed..8df0013 100644 --- a/internal/transport/turndial/turndial_test.go +++ b/internal/transport/turndial/turndial_test.go @@ -10,6 +10,7 @@ import ( "time" "github.com/pion/stun/v3" + "github.com/pion/turn/v5" ) func TestIsQuotaError(t *testing.T) { @@ -49,3 +50,86 @@ func TestOpen_HostOverrideApplied(t *testing.T) { t.Fatalf("expected dial error, got: %v", err) } } + +// Молчащий TURN: отмена ctx обязана прервать Allocate, а не ждать ретрансмиты pion. +func TestOpen_CancelInterruptsAllocate(t *testing.T) { + for _, udp := range []bool{false, true} { + t.Run(fmt.Sprintf("udp=%v", udp), func(t *testing.T) { + addr := silentTURN(t, udp) + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + peer := &net.UDPAddr{IP: net.ParseIP("1.2.3.4"), Port: 1} + + start := time.Now() + _, err := Open(ctx, Config{TransportUDP: udp}, peer, "u", "p", addr) + if d := time.Since(start); d > time.Second { + t.Fatalf("Open returned after %s, want prompt cancel", d) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Open = %v, want context.DeadlineExceeded", err) + } + }) + } +} + +func silentTURN(t *testing.T, udp bool) string { + t.Helper() + if udp { + pc, err := (&net.ListenConfig{}).ListenPacket(t.Context(), "udp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = pc.Close() }) + return pc.LocalAddr().String() + } + ln, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ln.Close() }) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + t.Cleanup(func() { _ = c.Close() }) + } + }() + return ln.Addr().String() +} + +// Успешный Allocate: хук отмены снят, Stream переживает последующую отмену ctx. +func TestOpen_AllocateThenCancelKeepsStream(t *testing.T) { + pc, err := (&net.ListenConfig{}).ListenPacket(t.Context(), "udp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + srv, err := turn.NewServer(turn.ServerConfig{ + Realm: "test", + AuthHandler: func(ra *turn.RequestAttributes) (string, []byte, bool) { + return ra.Username, turn.GenerateAuthKey(ra.Username, ra.Realm, "p"), true + }, + PacketConnConfigs: []turn.PacketConnConfig{{ + PacketConn: pc, + RelayAddressGenerator: &turn.RelayAddressGeneratorStatic{RelayAddress: net.ParseIP("127.0.0.1"), Address: "127.0.0.1"}, + }}, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = srv.Close() }) + + ctx, cancel := context.WithCancel(t.Context()) + peer := &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 1} + stream, err := Open(ctx, Config{TransportUDP: true}, peer, "u", "p", pc.LocalAddr().String()) + if err != nil { + t.Fatalf("Open: %v", err) + } + defer func() { _ = stream.Close() }() + cancel() + time.Sleep(50 * time.Millisecond) + if _, err := stream.Relay.WriteTo([]byte("x"), peer); err != nil { + t.Fatalf("relay closed by cancelled ctx: %v", err) + } +}