From b251b07e94b0bb1268e28d6694c4152aca15ae8d Mon Sep 17 00:00:00 2001 From: cppla Date: Sat, 3 Oct 2026 15:07:14 +0800 Subject: [PATCH] fix: join native dial resource cleanup on close --- docs/ARCHITECTURE.md | 17 + .../tunnel/native_quic_addr_resource_test.go | 195 ++++++++++ internal/tunnel/native_quic_addr_test.go | 83 +++++ internal/tunnel/native_tls_close_join_test.go | 352 ++++++++++++++++++ internal/tunnel/quic.go | 32 +- internal/tunnel/tls.go | 40 +- 6 files changed, 710 insertions(+), 9 deletions(-) create mode 100644 internal/tunnel/native_quic_addr_resource_test.go create mode 100644 internal/tunnel/native_quic_addr_test.go create mode 100644 internal/tunnel/native_tls_close_join_test.go diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index ff31948..69c0ee0 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -99,6 +99,23 @@ barrier. Canceling one caller still ends only its wait, not the shared physical attempt. An operation that ignores cancellation can delay closure, and best-effort observability callbacks remain outside this join guarantee. +The native client resolves the relay endpoint before asking the QUIC library to +allocate its owned UDP transport. Address failures therefore do not retain UDP +sockets until garbage collection. The numeric endpoint preserves mapped IPv6 +addresses and zones; the already configured TLS ServerName still owns certificate +verification. Successful connections retain the library's single-use transport +and socket cleanup behavior. + +The native TLS client closes its setup registration gate before cancellation and +waits for in-progress TCP/TLS/protocol setup to dispose of unreturned connections. +Concurrent client `Close` calls share the completion barrier. An application +closing an already returned stream remains registered until its socket close +finishes, so client closure also joins that close. This does not bind successfully +returned streams to their caller's setup context or change TCP half-close behavior. +An operation or TLS callback that ignores cancellation can delay closure. +TLS callbacks must not synchronously close the same client from inside a setup +that client closure needs to join. + ## Web-cover TCP flow The web-cover server binds TCP and UDP on the same numeric port. Its TCP side diff --git a/internal/tunnel/native_quic_addr_resource_test.go b/internal/tunnel/native_quic_addr_resource_test.go new file mode 100644 index 0000000..b268b88 --- /dev/null +++ b/internal/tunnel/native_quic_addr_resource_test.go @@ -0,0 +1,195 @@ +//go:build linux || darwin + +package tunnel + +import ( + "context" + "crypto/tls" + "errors" + "net" + "os" + "os/exec" + "runtime" + "runtime/debug" + "strings" + "syscall" + "testing" + "time" +) + +const ( + nativeQUICAddrResourceChild = "AUTOCAR_NATIVE_QUIC_ADDR_RESOURCE_CHILD" + nativeQUICAddrResourceScanLimit = 4096 +) + +// The child temporarily disables automatic GC, not the parent test process. +// This distinguishes synchronous ownership from eventual netFD finalization; +// it does not claim that the old error path permanently leaks descriptors. +func TestNativeQUICAddressErrorDoesNotRetainUDPDescriptors(t *testing.T) { + if os.Getenv(nativeQUICAddrResourceChild) == "1" { + nativeQUICAddrResourceChildTest(t) + return + } + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], + "-test.run=^TestNativeQUICAddressErrorDoesNotRetainUDPDescriptors$", + "-test.count=1", "-test.v", "-test.timeout=6s") + cmd.Env = append(os.Environ(), nativeQUICAddrResourceChild+"=1") + out, err := cmd.CombinedOutput() + for _, line := range strings.Split(strings.TrimSuffix(string(out), "\n"), "\n") { + t.Logf("isolated child | %s", line) + } + if ctx.Err() != nil { + t.Fatalf("isolated resource probe exceeded its budget: %v", ctx.Err()) + } + if err != nil { + t.Fatalf("isolated resource probe failed: %v", err) + } +} + +func nativeQUICAddrResourceChildTest(t *testing.T) { + const attempts = 32 + // Listen/Close is local socket allocation only; it warms runtime netpoll + // before counting descriptors and sends no packet to any destination. + warm, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero}) + if err != nil { + t.Fatal(err) + } + raw, rawErr := warm.SyscallConn() + var warmFD uintptr + var warmAddress syscall.Sockaddr + if rawErr == nil { + var addressErr error + rawErr = raw.Control(func(fd uintptr) { + warmFD = fd + warmAddress, addressErr = syscall.Getsockname(int(fd)) + }) + if rawErr == nil { + rawErr = addressErr + } + } + if err := warm.Close(); err != nil { + t.Fatal(err) + } + if rawErr != nil || warmFD >= nativeQUICAddrResourceScanLimit { + t.Fatalf("warm socket outside the bounded descriptor scan: fd=%d err=%v", warmFD, rawErr) + } + switch warmAddress.(type) { + case *syscall.SockaddrInet4, *syscall.SockaddrInet6: + default: + t.Fatalf("warm UDP socket is not an INET address: %T", warmAddress) + } + runtime.GC() + previousGC := debug.SetGCPercent(-1) + defer debug.SetGCPercent(previousGC) + before := nativeQUICAddrResourceSnapshot(t) + t.Logf("automatic_gc=disabled attempts=%d before_open=%d before_udp=%d", attempts, before.open, before.udp) + + for i := 0; i < attempts; i++ { + // The numeric out-of-range port is rejected before IP lookup. No DNS + // query, QUIC handshake, or connection to the valid TCP target occurs. + client, err := NewClient(ClientConfig{ + ServerAddress: "127.0.0.1:65536", Token: testToken, + TLSConfig: &tls.Config{ServerName: "127.0.0.1"}, + QUICDialTimeout: time.Second, + }) + if err != nil { + t.Fatalf("attempt %d constructor rejected the deferred address case: %v", i, err) + } + dialCtx, cancelDial := context.WithTimeout(context.Background(), time.Second) + conn, dialErr := client.DialContext(dialCtx, "tcp", "127.0.0.1:1") + cancelDial() + if conn != nil { + _ = conn.Close() + t.Errorf("attempt %d returned a connection for an invalid server port", i) + } + if err := client.Close(); err != nil { + t.Errorf("attempt %d client Close: %v", i, err) + } + var addressErr *net.AddrError + if !errors.As(dialErr, &addressErr) || addressErr.Err != "invalid port" || addressErr.Addr != "65536" { + t.Errorf("attempt %d did not preserve the numeric address error: %T %v", i, dialErr, dialErr) + } + } + // Close has joined every default physical attempt. Any UDP descriptor + // still retained here cannot be attributed to an unfinished dial worker. + afterClose := nativeQUICAddrResourceSnapshot(t) + retained := afterClose.udp - before.udp + t.Logf("after_client_close_open=%d after_client_close_udp=%d retained_udp=%d", afterClose.open, afterClose.udp, retained) + if retained != 0 { + t.Errorf("client Close left %d UDP descriptors retained after %d completed address failures; cleanup must not depend on GC", retained, attempts) + } + + // Preserve this diagnostic even when the primary no-retention assertion + // fails: a subsequent GC may reclaim the old path's unreachable netFDs. + runtime.GC() + gcDeadline := time.Now().Add(2 * time.Second) + afterGC := nativeQUICAddrResourceSnapshot(t) + for afterGC.udp != before.udp && time.Now().Before(gcDeadline) { + runtime.Gosched() + time.Sleep(5 * time.Millisecond) + afterGC = nativeQUICAddrResourceSnapshot(t) + } + t.Logf("after_explicit_gc_open=%d after_explicit_gc_udp=%d post_gc_udp_delta=%d", afterGC.open, afterGC.udp, afterGC.udp-before.udp) + if afterGC.udp != before.udp { + t.Errorf("explicit GC did not restore UDP descriptor count within the diagnostic budget: before=%d after=%d", before.udp, afterGC.udp) + } +} + +type nativeQUICAddrResources struct { + open int + udp int +} + +func nativeQUICAddrResourceSnapshot(t *testing.T) nativeQUICAddrResources { + t.Helper() + var limit syscall.Rlimit + if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &limit); err != nil { + t.Fatal(err) + } + // The isolated child has only runtime/stdio descriptors plus at most 32 + // sockets. Scan a fixed bounded range; do not raise or lower host limits. + bound := nativeQUICAddrResourceScanLimit + if limit.Cur < nativeQUICAddrResourceScanLimit { + bound = int(limit.Cur) + } + if bound < 128 { + t.Fatalf("descriptor limit too small for the bounded child probe: %d", bound) + } + var resources nativeQUICAddrResources + for fd := 0; fd < bound; fd++ { + _, _, errno := syscall.Syscall(syscall.SYS_FCNTL, uintptr(fd), uintptr(syscall.F_GETFD), 0) + if errno == syscall.EBADF { + continue + } + if errno != 0 { + t.Fatalf("F_GETFD(%d): %v", fd, errno) + } + resources.open++ + kind, err := syscall.GetsockoptInt(fd, syscall.SOL_SOCKET, syscall.SO_TYPE) + if err == nil && kind == syscall.SOCK_DGRAM { + address, err := syscall.Getsockname(fd) + // Explicit GC can run netFD finalizers between these read-only + // descriptor syscalls; a just-closed descriptor is not a failure. + if errors.Is(err, syscall.EBADF) { + continue + } + // SOCK_DGRAM does not imply INET UDP. Non-INET control or + // runtime descriptors may not support this address query; do + // not count them as UDP. The owned warm INET socket above + // verifies that its address query works without this filter. + if errors.Is(err, syscall.EOPNOTSUPP) || errors.Is(err, syscall.EAFNOSUPPORT) || errors.Is(err, syscall.ENOTSOCK) { + continue + } + if err != nil { + t.Fatalf("getsockname(%d): %v", fd, err) + } + switch address.(type) { + case *syscall.SockaddrInet4, *syscall.SockaddrInet6: + resources.udp++ + } + } + } + return resources +} diff --git a/internal/tunnel/native_quic_addr_test.go b/internal/tunnel/native_quic_addr_test.go new file mode 100644 index 0000000..74e2049 --- /dev/null +++ b/internal/tunnel/native_quic_addr_test.go @@ -0,0 +1,83 @@ +package tunnel + +import ( + "context" + "net/netip" + "testing" +) + +func TestResolveNativeQUICAddress(t *testing.T) { + for _, tc := range []struct { + name string + address string + want string + zone string + mapped bool + empty bool + }{ + {name: "ipv4", address: "127.0.0.1:443", want: "[::ffff:127.0.0.1]:443", mapped: true}, + {name: "ipv6", address: "[::1]:443", want: "[::1]:443"}, + {name: "ipv6_canonical", address: "[2001:DB8:0:0:0:0:0:1]:443", want: "[2001:db8::1]:443"}, + {name: "mapped_ipv6", address: "[::ffff:127.0.0.1]:443", want: "[::ffff:127.0.0.1]:443", mapped: true}, + {name: "ipv6_zone", address: "[fe80::1%audit-zone]:443", want: "[fe80::1%audit-zone]:443", zone: "audit-zone"}, + {name: "mapped_ipv6_zone", address: "[::ffff:127.0.0.1%audit-zone]:443", want: "[::ffff:127.0.0.1%audit-zone]:443", zone: "audit-zone", mapped: true}, + {name: "empty_host", address: ":443", want: ":443", empty: true}, + {name: "empty_address", address: "", want: ":0", empty: true}, + // This uses a literal IP and the standard local UDP service name; + // neither an external hostname nor an interface is resolved. + {name: "named_port", address: "127.0.0.1:domain", want: "[::ffff:127.0.0.1]:53", mapped: true}, + {name: "ipv4_zero_port", address: "127.0.0.1:0", want: "[::ffff:127.0.0.1]:0", mapped: true}, + {name: "ipv6_zero_port", address: "[::1]:0", want: "[::1]:0"}, + {name: "empty_host_zero_port", address: ":0", want: ":0", empty: true}, + } { + t.Run(tc.name, func(t *testing.T) { + got, err := resolveNativeQUICAddress(tc.address) + if err != nil { + t.Fatal(err) + } + if got != tc.want { + t.Errorf("numeric authority = %q, want %q", got, tc.want) + } + if tc.empty { + return + } + // Literal parsing proves that the second DialAddr resolution need + // not perform DNS. In particular IPv4%zone is not a valid literal. + endpoint, err := netip.ParseAddrPort(got) + if err != nil { + t.Fatalf("resolved authority is not a literal endpoint: %v", err) + } + if endpoint.Addr().Zone() != tc.zone || endpoint.Addr().Is4In6() != tc.mapped { + t.Errorf("family/zone changed: endpoint=%v mapped=%t zone=%q", endpoint, endpoint.Addr().Is4In6(), endpoint.Addr().Zone()) + } + }) + } + for _, tc := range []struct{ name, address string }{ + {"missing_port", "127.0.0.1"}, + {"unbracketed_ipv6", "::1:443"}, + {"high_port", "127.0.0.1:65536"}, + {"negative_port", "127.0.0.1:-1"}, + {"extra_port_separator", "127.0.0.1:443:1"}, + {"missing_ipv6_bracket", "[::1:443"}, + } { + t.Run(tc.name, func(t *testing.T) { + got, err := resolveNativeQUICAddress(tc.address) + if err == nil || got != "" { + t.Errorf("malformed address resolved: got=%q err=%v", got, err) + } + }) + } +} + +func TestDialNativeQUICAddrRejectsNilTLS(t *testing.T) { + // The default NewClient path always supplies a verifying TLS config. + // This private-wrapper boundary is rejected before allocating a socket, + // without dialing or observing any external peer. + conn, err := dialNativeQUICAddr(context.Background(), "127.0.0.1:1", nil, nil) + if conn != nil || err == nil || err.Error() != "quic: tls.Config not set" { + if conn != nil { + _ = conn.CloseWithError(applicationShutdown, "unexpected nil TLS result") + } + t.Fatalf("nil TLS result: conn=%v err=%v", conn, err) + } +} diff --git a/internal/tunnel/native_tls_close_join_test.go b/internal/tunnel/native_tls_close_join_test.go new file mode 100644 index 0000000..33ce1e8 --- /dev/null +++ b/internal/tunnel/native_tls_close_join_test.go @@ -0,0 +1,352 @@ +package tunnel + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/protocol" +) + +// This context preserves the actual caller's values and cancellation channel. +// The only scheduling gate is its first Deadline return after trusted TLS +// verification. In the pinned implementation this occurs in openProtocol, +// after the TLS handshake and before that exchange's cancellation watcher. +type nativeTLSSetupGateContext struct { + context.Context + armed atomic.Bool + once sync.Once + entered chan struct{} + release <-chan struct{} +} + +func (c *nativeTLSSetupGateContext) Deadline() (time.Time, bool) { + if c.armed.Load() { + c.once.Do(func() { + close(c.entered) + select { + case <-c.release: + case <-c.Context.Done(): + } + }) + } + return c.Context.Deadline() +} + +func TestNativeTLSCloseJoinsUnregisteredSetup(t *testing.T) { + serverTLS, clientTLS := testTLSConfigs(t) + serverTLS = mustServerTLSConfig(t, serverTLS) + listener, err := net.ListenTCP("tcp4", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := listener.SetDeadline(time.Now().Add(2 * time.Second)); err != nil { + _ = listener.Close() + t.Fatal(err) + } + release := make(chan struct{}) + var releaseOnce sync.Once + openGate := func() { releaseOnce.Do(func() { close(release) }) } + callerCtx := &nativeTLSSetupGateContext{Context: ctx, entered: make(chan struct{}), release: release} + clientTLS.VerifyConnection = func(state tls.ConnectionState) error { + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != protocol.ALPN || len(state.VerifiedChains) == 0 { + return errors.New("test: missing trusted TLS 1.3/native ALPN") + } + callerCtx.armed.Store(true) + return nil + } + client, err := NewTLSClient(TLSClientConfig{ServerAddress: listener.Addr().String(), Token: testToken, + TLSConfig: clientTLS, HandshakeTimeout: 2 * time.Second, DialTimeout: 2 * time.Second}) + if err != nil { + _ = listener.Close() + t.Fatal(err) + } + type peerResult struct { + conn *tls.Conn + err error + } + peerReady := make(chan peerResult, 1) + peerRead := make(chan error, 1) + peerEOF := make(chan error, 1) + readStart := make(chan struct{}) + var readOnce sync.Once + startRead := func() { readOnce.Do(func() { close(readStart) }) } + peerDone := make(chan struct{}) + go func() { + defer close(peerDone) + raw, err := listener.AcceptTCP() + if err != nil { + peerReady <- peerResult{err: err} + return + } + defer raw.Close() + conn := tls.Server(raw, serverTLS) + err = conn.HandshakeContext(ctx) + peerReady <- peerResult{conn: conn, err: err} + if err != nil { + return + } + select { + case <-readStart: + case <-ctx.Done(): + return + } + _ = conn.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) + var one [1]byte + // Both TLS handshakes have completed, and the caller's Deadline gate + // still precedes its first protocol write. Observe this empty raw socket + // without making a TLS parser timeout that would be unsafe to recover. + n, err := conn.NetConn().Read(one[:]) + if n != 0 { + err = fmt.Errorf("test: gated pre-request phase unexpectedly read %d raw bytes", n) + } + peerRead <- err + // The still-live socket is allowed while Close itself remains pending. + // Do not manufacture peer closure: after setup is released, observe the + // client's real socket Close by draining raw TCP to EOF. A canceled TLS + // write can corrupt its record sequence, so this ownership test does not + // require a graceful TLS close_notify after interrupted protocol setup. + select { + case <-release: + case <-ctx.Done(): + return + } + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + _, err = io.Copy(io.Discard, conn.NetConn()) + peerEOF <- err + }() + callerDone := make(chan struct{}) + callerResult := make(chan error, 1) + go func() { + defer close(callerDone) + conn, err := client.DialContext(callerCtx, "tcp", listener.Addr().String()) + if conn != nil { + _ = conn.Close() + if err == nil { + err = errors.New("test: unexpected stream handoff after client Close") + } + } + callerResult <- err + }() + closeDone := [2]chan struct{}{make(chan struct{}), make(chan struct{})} + closeResults := make(chan error, 2) + var closeStarted [2]bool + var peer *tls.Conn + t.Cleanup(func() { + openGate() + startRead() + cancel() + _ = listener.Close() + if peer != nil { + _ = peer.NetConn().Close() + } + _ = client.Close() + nativeTLSCloseJoinWait(t, callerDone, "setup caller cleanup") + for index, started := range closeStarted { + if started { + nativeTLSCloseJoinWait(t, closeDone[index], "Close cleanup") + } + } + nativeTLSCloseJoinWait(t, peerDone, "peer cleanup") + }) + select { + case <-callerCtx.entered: + case <-callerDone: + t.Fatalf("setup did not reach post-TLS gate: %v", <-callerResult) + case <-ctx.Done(): + t.Fatal("post-TLS setup gate not reached") + } + select { + case ready := <-peerReady: + peer = ready.conn + if ready.err != nil { + t.Fatal(ready.err) + } + case <-ctx.Done(): + t.Fatal("real TLS peer handshake did not complete") + } + state := peer.ConnectionState() + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != protocol.ALPN { + t.Fatalf("actual peer TLS=%#x ALPN=%q", state.Version, state.NegotiatedProtocol) + } + client.mu.Lock() + registered := len(client.conns) + client.mu.Unlock() + if registered != 0 || ctx.Err() != nil { + t.Fatalf("gate retained registered=%d caller=%v", registered, ctx.Err()) + } + for index := range closeDone { + closeStarted[index] = true + go func(index int) { defer close(closeDone[index]); closeResults <- client.Close() }(index) + if index == 0 && !nativeTLSCloseJoinWait(t, client.ctx.Done(), "first Close closed gate") { + return + } + } + for index := range closeDone { + select { + case <-closeDone[index]: + t.Errorf("Close %d returned before the actual unregistered TLS setup completed", index) + case <-time.After(75 * time.Millisecond): + } + } + startRead() + select { + case err := <-peerRead: + var timeout net.Error + if !errors.As(err, &timeout) || !timeout.Timeout() { + t.Errorf("gated actual TLS socket read=%v, want a live-socket timeout", err) + } else { + t.Log("actual TLS peer remained live while post-handshake setup return was gated") + } + case <-ctx.Done(): + t.Error("actual peer closure was not observed") + } + if ctx.Err() != nil { + t.Error("caller/fixture lifetime canceled before independent cleanup") + } + openGate() + if !nativeTLSCloseJoinWait(t, callerDone, "setup caller") { + return + } + if err := <-callerResult; err == nil { + t.Error("closed client returned a successful setup") + } + for index := range closeDone { + if !nativeTLSCloseJoinWait(t, closeDone[index], "Close completion") { + return + } + if err := <-closeResults; err != nil { + t.Errorf("Close=%v", err) + } + } + if !nativeTLSCloseJoinWait(t, peerDone, "real peer completion") { + return + } + select { + case err := <-peerEOF: + if err != nil { + t.Errorf("actual TCP peer did not drain to EOF after setup cleanup: %v", err) + } + default: + t.Error("real peer exited without recording post-release EOF") + } + if _, err := client.DialContext(ctx, "tcp", listener.Addr().String()); !errors.Is(err, net.ErrClosed) { + t.Errorf("future dial=%v, want net.ErrClosed", err) + } + t.Log("trusted TLS1.3/native ALPN completed; setup gate preceded native request watcher, not an authenticated target exchange") +} + +func TestNativeTLSCloseJoinsApplicationClose(t *testing.T) { + f := newRelayOwnerFixture(t, "tls", "stalled", false) + tracked := f.conn.(*trackedTLSConn) + client := tracked.owner + raw := tracked.NetConn().(*net.TCPConn) + control, err := raw.SyscallConn() + if err != nil { + t.Fatal(err) + } + release := make(chan struct{}) + var releaseOnce sync.Once + openGate := func() { releaseOnce.Do(func() { close(release) }) } + entered, controlDone := make(chan struct{}), make(chan struct{}) + controlResult := make(chan error, 1) + appDone, ownerDone := make(chan struct{}), make(chan struct{}) + appResult, ownerResult := make(chan error, 1), make(chan error, 1) + var appStarted, ownerStarted bool + t.Cleanup(func() { + openGate() + nativeTLSCloseJoinWait(t, controlDone, "raw FD control cleanup") + if appStarted { + nativeTLSCloseJoinWait(t, appDone, "application Close cleanup") + } + if ownerStarted { + nativeTLSCloseJoinWait(t, ownerDone, "client Close cleanup") + } + }) + go func() { + defer close(controlDone) + controlResult <- control.Control(func(uintptr) { + // This is an actual socket FD reference, not a fake Close result. + // netFD.Close cannot finish destroying it before Control returns. + close(entered) + <-release + }) + }() + if !nativeTLSCloseJoinWait(t, entered, "actual FD Control entry") { + return + } + appStarted = true + go func() { defer close(appDone); appResult <- tracked.Close() }() + deadline, ticker := time.NewTimer(time.Second), time.NewTicker(10*time.Millisecond) + defer deadline.Stop() + defer ticker.Stop() + for { + // No extra reader consumes tunnel bytes. The real TCP API witnesses + // Close's closed-poller state while the held FD reference prevents return. + if err := raw.SetReadDeadline(time.Now().Add(time.Second)); errors.Is(err, net.ErrClosed) { + break + } else if err != nil { + t.Fatal(err) + } + select { + case <-ticker.C: + case <-deadline.C: + t.Fatal("actual raw Close did not reach its FD-reference wait") + } + } + select { + case <-appDone: + t.Fatal("actual application Close completed before releasing its FD reference") + default: + } + ownerStarted = true + go func() { defer close(ownerDone); ownerResult <- client.Close() }() + if !nativeTLSCloseJoinWait(t, client.ctx.Done(), "owner Close closed gate") { + return + } + select { + case <-ownerDone: + t.Error("client Close returned before an actual application Close finished") + case <-time.After(75 * time.Millisecond): + } + openGate() + if !nativeTLSCloseJoinWait(t, controlDone, "raw FD Control completion") || + !nativeTLSCloseJoinWait(t, appDone, "application Close completion") || + !nativeTLSCloseJoinWait(t, ownerDone, "client Close completion") { + return + } + for name, result := range map[string]<-chan error{"control": controlResult, "application": appResult, "owner": ownerResult} { + if err := <-result; err != nil { + t.Errorf("%s=%v", name, err) + } + } + client.mu.Lock() + registered := len(client.conns) + client.mu.Unlock() + if registered != 0 { + t.Errorf("completed Close retains %d registered streams", registered) + } + t.Log("actual authenticated TLS target marker/ack preceded app Close; real FD-reference wait was independently released") +} + +func nativeTLSCloseJoinWait(t *testing.T, done <-chan struct{}, phase string) bool { + t.Helper() + select { + case <-done: + return true + case <-time.After(2 * time.Second): + t.Errorf("%s did not join", phase) + return false + } +} + +var _ context.Context = (*nativeTLSSetupGateContext)(nil) diff --git a/internal/tunnel/quic.go b/internal/tunnel/quic.go index 76833df..1ff9e14 100644 --- a/internal/tunnel/quic.go +++ b/internal/tunnel/quic.go @@ -571,7 +571,7 @@ func NewClient(config ClientConfig) (*Client, error) { token: config.Token, tlsConfig: tlsConfig, quicConfig: hardenedQUICClientConfig(config.QUICConfig), - dialQUIC: quic.DialAddr, + dialQUIC: dialNativeQUICAddr, handshakeTimeout: handshakeTimeout, dialTimeout: dialTimeout, primaryTimeout: primaryTimeout, @@ -613,6 +613,36 @@ func NewClient(config ClientConfig) (*Client, error) { // NewQUICClient is an explicit alias for NewClient. func NewQUICClient(config ClientConfig) (*Client, error) { return NewClient(config) } +// Resolve before DialAddr allocates its owned UDP socket. The pinned library's +// address-resolution error path otherwise leaves that socket to GC. NewClient +// already fixes the verifying TLS ServerName, so using a numeric endpoint here +// changes neither certificate identity nor the single-use transport ownership. +func dialNativeQUICAddr(ctx context.Context, address string, tlsConfig *tls.Config, config *quic.Config) (*quic.Conn, error) { + numericAddress, err := resolveNativeQUICAddress(address) + if err != nil { + return nil, err + } + if tlsConfig == nil { + return nil, errors.New("quic: tls.Config not set") + } + return quic.DialAddr(ctx, numericAddress, tlsConfig, config) +} + +func resolveNativeQUICAddress(address string) (string, error) { + resolved, err := net.ResolveUDPAddr("udp", address) + if err != nil { + return "", err + } + // UDPAddr.String folds mapped IPv6 into IPv4, which cannot carry a zone + // through a second resolution. AddrPort preserves both the family and zone. + if endpoint := resolved.AddrPort(); endpoint.IsValid() { + return endpoint.String(), nil + } + // Preserve the existing unspecified-host form when an explicit TLS + // ServerName allowed it through client configuration validation. + return resolved.String(), nil +} + func hardenedQUICClientConfig(input *quic.Config) *quic.Config { var cfg *quic.Config if input == nil { diff --git a/internal/tunnel/tls.go b/internal/tunnel/tls.go index 1e0fcb3..a11c192 100644 --- a/internal/tunnel/tls.go +++ b/internal/tunnel/tls.go @@ -305,6 +305,9 @@ type TLSClient struct { closed bool selected bool conns map[*trackedTLSConn]struct{} + // Setup registration and Close's gate share mu, so no Add can race Wait. + setups sync.WaitGroup + closeDone chan struct{} } // SelectedTransport identifies the concrete path for benchmark telemetry. @@ -350,6 +353,7 @@ func NewTLSClient(config TLSClientConfig) (*TLSClient, error) { ctx: ctx, cancel: cancel, conns: make(map[*trackedTLSConn]struct{}), + closeDone: make(chan struct{}), }, nil } @@ -366,15 +370,25 @@ func (c *TLSClient) DialContext(ctx context.Context, network, address string) (n return nil, err } c.mu.Lock() - closed := c.closed - c.mu.Unlock() - if closed { + if c.closed { + c.mu.Unlock() return nil, net.ErrClosed } + c.setups.Add(1) + c.mu.Unlock() + defer c.setups.Done() dialCtx, cancel := context.WithCancel(ctx) - stop := context.AfterFunc(c.ctx, cancel) - defer stop() + cancelDone := make(chan struct{}) + stop := context.AfterFunc(c.ctx, func() { + cancel() + close(cancelDone) + }) + defer func() { + if !stop() { + <-cancelDone + } + }() defer cancel() raw, err := c.dialer.DialContext(dialCtx, "tcp", c.address) if err != nil { @@ -410,23 +424,31 @@ func (c *TLSClient) DialContext(ctx context.Context, network, address string) (n return tracked, nil } -// Close prevents future dials and closes every active fallback stream. +// Close prevents future dials, closes active streams and joins in-flight setup. +// Concurrent calls wait for the same cleanup to finish. TLS callbacks must +// return themselves and must not synchronously Close their own setup client. func (c *TLSClient) Close() error { c.mu.Lock() if c.closed { + done := c.closeDone c.mu.Unlock() + <-done return nil } c.closed = true - c.cancel() connections := make([]*trackedTLSConn, 0, len(c.conns)) for conn := range c.conns { connections = append(connections, conn) } c.mu.Unlock() + c.cancel() for _, conn := range connections { _ = conn.Close() } + // An unregistered late result is still owned by its setup until cleanup + // returns. Never hold mu while joining setup or actually closing a socket. + c.setups.Wait() + close(c.closeDone) return nil } @@ -445,7 +467,9 @@ type trackedTLSConn struct { func (c *trackedTLSConn) Close() error { var err error c.once.Do(func() { - c.owner.remove(c) + // Retain registry ownership until the actual Close returns, allowing a + // simultaneous client Close to find this stream and join once.Do. + defer c.owner.remove(c) err = c.Conn.Close() }) return err