From 064875d32f33c15f0c1fd34a38be6f3e752f2e87 Mon Sep 17 00:00:00 2001 From: cppla Date: Sat, 3 Oct 2026 12:18:52 +0800 Subject: [PATCH 1/2] Join stalled TCP destinations on native relay shutdown --- docs/ARCHITECTURE.md | 9 + internal/tunnel/common.go | 18 + internal/tunnel/quic.go | 2 +- internal/tunnel/relay_owner_shutdown_test.go | 568 +++++++++++++++++++ internal/tunnel/tls.go | 4 +- 5 files changed, 599 insertions(+), 2 deletions(-) create mode 100644 internal/tunnel/relay_owner_shutdown_test.go diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index de38051..3cedaf5 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -60,6 +60,15 @@ untracked Go source for the known disallowed proxy application-module import. directions. Only then does raw byte relay begin. 6. EOF is half-closed when possible; hard errors abort both copy directions. +Native QUIC and TLS fallback relays also own destination cleanup when their +physical connection or server is shut down. Cancellation closes the TCP target +to interrupt idle reads as well as pending writes, then joins both relay pumps +and any in-progress close callback before releasing stream admission. The +owner is the physical QUIC connection or TLS server lifetime, never the dial +timer or QUIC send-side stream context. An ordinary response FIN therefore +still permits upload, and client half-close still permits a delayed response. +This does not impose a new timeout on healthy idle destination reads. + QUIC streams are independent. Canceling a slow open or one flow does not close sibling flows. A per-connection pacer sees aggregate QUIC statistics, so concurrent streams share one path estimate rather than each overestimating the diff --git a/internal/tunnel/common.go b/internal/tunnel/common.go index 19d430f..0a470d1 100644 --- a/internal/tunnel/common.go +++ b/internal/tunnel/common.go @@ -164,6 +164,10 @@ type streamRequestOptions struct { Response protocol.Response Relay deadlineConn Handled bool + // RelayOwner ends the whole relay, not just dialing or one stream direction. + // Its owner must also interrupt ingress I/O when canceled. A QUIC send-side + // stream context is unsuitable: normal response FIN cancels that context. + RelayOwner context.Context } type streamRequestHandler func(context.Context, deadlineConn, protocol.Request) streamRequestOptions @@ -235,6 +239,20 @@ func (s *serverCore) handleStream( } upstream = s.boundDestinationWrites(upstream) defer upstream.Close() + if options.RelayOwner != nil { + closed := make(chan struct{}) + stop := context.AfterFunc(options.RelayOwner, func() { + _ = upstream.Close() + close(closed) + }) + defer func() { + // Stop does not join a callback that has already started. Keep the + // stream lease until its owned destination Close has returned. + if !stop() { + <-closed + } + }() + } options.Response.Status = protocol.StatusOK if err := protocol.WriteResponse(stream, options.Response); err != nil { return diff --git a/internal/tunnel/quic.go b/internal/tunnel/quic.go index 8bab686..29c8813 100644 --- a/internal/tunnel/quic.go +++ b/internal/tunnel/quic.go @@ -363,7 +363,7 @@ func (s *QUICServer) serveConnection(conn *quic.Conn, sourceKey string) { if request.Network == protocol.NetworkUDP { return datagrams.handleRequest(ctx, requestStream, request, response) } - return streamRequestOptions{Response: response, Relay: requestStream} + return streamRequestOptions{Response: response, Relay: requestStream, RelayOwner: conn.Context()} } s.core.handleStream(stream.Context(), wrapped, markAuthenticated, handler) }() diff --git a/internal/tunnel/relay_owner_shutdown_test.go b/internal/tunnel/relay_owner_shutdown_test.go new file mode 100644 index 0000000..16eef1c --- /dev/null +++ b/internal/tunnel/relay_owner_shutdown_test.go @@ -0,0 +1,568 @@ +package tunnel + +import ( + "bytes" + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/protocol" + "github.com/cppla/autocar/internal/transport" +) + +const relayOwnerMarker = "authenticated-relay-owner-marker" +const relayOwnerAck = "owned-target-ack" + +// The target, relay, and client all use actual owned loopback sockets. Only +// target application progress and the return of its first Close are gated; +// neither gate manufactures a transport error or consumes tunnel bytes. +type relayOwnerFixture struct { + t *testing.T + mode, targetMode string + listener *net.TCPListener + mu sync.Mutex + closing bool + origin *net.TCPConn + destination *relayOwnerDestination + workers []<-chan struct{} + release, closeRelease chan struct{} + releaseOnce, closeReleaseOnce sync.Once + originEOF, originDone chan struct{} + originErr error + response, tail []byte + dialDeadline time.Time + dials atomic.Int32 + tlsServer *TLSServer + quicServer *QUICServer + closeServer func() error + closeClient func() error + serveCancel context.CancelFunc + serveContext context.Context + serveDone chan struct{} + serveErr error + conn net.Conn +} + +type relayOwnerDestination struct { + *net.TCPConn + closeRelease <-chan struct{} + closed, closeEntered, halfClosed chan struct{} + closeDone chan struct{} + closeStarted atomic.Bool + halfOnce sync.Once + closeCalls atomic.Int32 + closeErr error +} + +func (c *relayOwnerDestination) Close() error { + if c.closeStarted.CompareAndSwap(false, true) { + defer close(c.closeDone) + c.closeCalls.Add(1) + c.closeErr = c.TCPConn.Close() + close(c.closed) + close(c.closeEntered) + <-c.closeRelease + return c.closeErr + } + // Subsequent calls deliberately do not join the first return gate. The + // server must join its owner callback, not accidentally rely on this fixture + // making the ordinary deferred target Close wait for that callback. + return nil +} + +func (c *relayOwnerDestination) CloseWrite() error { + err := c.TCPConn.CloseWrite() + if err == nil { + c.halfOnce.Do(func() { close(c.halfClosed) }) + } + return err +} + +func newRelayOwnerFixture(t *testing.T, mode, targetMode string, holdClose bool) *relayOwnerFixture { + t.Helper() + f := &relayOwnerFixture{ + t: t, mode: mode, targetMode: targetMode, + release: make(chan struct{}), closeRelease: make(chan struct{}), + originEOF: make(chan struct{}), originDone: make(chan struct{}), serveDone: make(chan struct{}), + response: bytes.Repeat([]byte("complete-response-"), 8192), + tail: bytes.Repeat([]byte("late-upload-tail"), 4096), + } + t.Cleanup(f.cleanup) + if !holdClose { + f.releaseClose() + } + listener, err := net.ListenTCP("tcp4", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + f.listener = listener + if err := listener.SetDeadline(time.Now().Add(4 * time.Second)); err != nil { + t.Fatal(err) + } + f.register(f.originDone) + go f.serveOrigin() + + dialer := transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + if network != "tcp" || address != listener.Addr().String() { + return nil, fmt.Errorf("unowned destination %q/%q", network, address) + } + f.dials.Add(1) + raw, err := (&net.Dialer{}).DialContext(ctx, "tcp4", address) + if err != nil { + return nil, err + } + destination := &relayOwnerDestination{TCPConn: raw.(*net.TCPConn), closeRelease: f.closeRelease, + closed: make(chan struct{}), closeEntered: make(chan struct{}), halfClosed: make(chan struct{}), closeDone: make(chan struct{})} + f.register(destination.closeDone) + f.mu.Lock() + f.destination = destination + f.dialDeadline, _ = ctx.Deadline() + closing := f.closing + f.mu.Unlock() + if closing { + _ = destination.Close() + return nil, net.ErrClosed + } + return destination, nil + }) + serverTLS, clientTLS := testTLSConfigs(t) + f.serveContext, f.serveCancel = context.WithCancel(context.Background()) + var client transport.Dialer + var serve func(context.Context) error + if mode == "tls" { + server, err := ListenTLS(TLSServerConfig{Address: "127.0.0.1:0", Token: testToken, + TLSConfig: serverTLS, Dialer: dialer, DialTimeout: 100 * time.Millisecond, + MaxConcurrentStreams: 1, MaxClientConnections: 1}) + if err != nil { + t.Fatal(err) + } + f.tlsServer, f.closeServer, serve = server, server.Close, server.Serve + tlsClient, err := NewTLSClient(TLSClientConfig{ServerAddress: server.Addr().String(), Token: testToken, TLSConfig: clientTLS}) + if err != nil { + t.Fatal(err) + } + client, f.closeClient = tlsClient, tlsClient.Close + } else { + server, err := ListenQUIC(QUICServerConfig{Address: "127.0.0.1:0", Token: testToken, + TLSConfig: serverTLS, Dialer: dialer, DialTimeout: 100 * time.Millisecond, + MaxConcurrentStreams: 1, MaxConnections: 1, MaxClientConnections: 1}) + if err != nil { + t.Fatal(err) + } + f.quicServer, f.closeServer, serve = server, server.Close, server.Serve + quicClient, err := NewClient(ClientConfig{ServerAddress: server.Addr().String(), Token: testToken, TLSConfig: clientTLS}) + if err != nil { + t.Fatal(err) + } + client, f.closeClient = quicClient, quicClient.Close + } + f.register(f.serveDone) + go func() { defer close(f.serveDone); f.serveErr = serve(f.serveContext) }() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + f.conn, err = client.DialContext(ctx, "tcp", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + if err := f.conn.SetDeadline(time.Now().Add(3 * time.Second)); err != nil { + t.Fatal(err) + } + if mode == "tls" { + state := f.conn.(*trackedTLSConn).ConnectionState() + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != protocol.ALPN { + t.Fatalf("TLS security state=%+v", state) + } + } else { + quicClient := client.(*Client) + quicClient.mu.Lock() + physical := quicClient.conn + quicClient.mu.Unlock() + if physical == nil { + t.Fatal("missing actual QUIC connection") + } + state := physical.ConnectionState() + if state.TLS.Version != tls.VersionTLS13 || state.TLS.NegotiatedProtocol != protocol.ALPN || state.Used0RTT { + t.Fatalf("QUIC security state=%+v", state) + } + } + if _, err := io.Copy(f.conn, bytes.NewBufferString(relayOwnerMarker)); err != nil { + t.Fatal(err) + } + ack := make([]byte, len(relayOwnerAck)) + if _, err := io.ReadFull(f.conn, ack); err != nil || string(ack) != relayOwnerAck { + t.Fatalf("actual relay ack=%q, %v", ack, err) + } + f.assertSlots(1, true) + if f.dials.Load() != 1 { + t.Fatalf("actual authenticated target dials=%d, want 1", f.dials.Load()) + } + t.Logf("%s: verified TLS1.3/%s, actual authenticated target marker/ack", mode, protocol.ALPN) + return f +} + +func (f *relayOwnerFixture) serveOrigin() { + defer close(f.originDone) + conn, err := f.listener.AcceptTCP() + if err != nil { + f.originErr = err + return + } + defer conn.Close() + f.mu.Lock() + f.origin = conn + closing := f.closing + f.mu.Unlock() + if closing { + return + } + if err := conn.SetDeadline(time.Now().Add(4 * time.Second)); err != nil { + f.originErr = err + return + } + marker := make([]byte, len(relayOwnerMarker)) + if _, err := io.ReadFull(conn, marker); err != nil || string(marker) != relayOwnerMarker { + f.originErr = fmt.Errorf("origin marker=%q: %v", marker, err) + return + } + if _, err := conn.Write([]byte(relayOwnerAck)); err != nil { + f.originErr = err + return + } + if f.targetMode == "response_first" { + if _, err := io.Copy(conn, bytes.NewReader(f.response)); err != nil { + f.originErr = err + return + } + if err := conn.CloseWrite(); err != nil { + f.originErr = err + return + } + } + upload, err := io.ReadAll(conn) + if err != nil { + f.originErr = err + return + } + if f.targetMode == "response_first" && !bytes.Equal(upload, f.tail) { + f.originErr = fmt.Errorf("complete late tail mismatch: got %d, want %d bytes", len(upload), len(f.tail)) + return + } + if f.targetMode != "response_first" && len(upload) != 0 { + f.originErr = fmt.Errorf("unexpected target upload %d bytes", len(upload)) + return + } + close(f.originEOF) + if f.targetMode == "response_first" { + return + } + <-f.release + if f.targetMode == "delayed" { + if _, err := io.Copy(conn, bytes.NewReader(f.response)); err != nil { + f.originErr = err + return + } + f.originErr = conn.CloseWrite() + } +} + +func (f *relayOwnerFixture) register(done <-chan struct{}) { + f.mu.Lock() + f.workers = append(f.workers, done) + f.mu.Unlock() +} + +func (f *relayOwnerFixture) wait(done <-chan struct{}, label string, budget time.Duration) bool { + f.t.Helper() + timer := time.NewTimer(budget) + defer timer.Stop() + select { + case <-done: + return true + case <-timer.C: + f.t.Errorf("%s did not complete within %s", label, budget) + return false + } +} + +func (f *relayOwnerFixture) require(done <-chan struct{}, label string) { + f.t.Helper() + if !f.wait(done, label, 2*time.Second) { + f.t.FailNow() + } +} + +func (f *relayOwnerFixture) releaseTarget() { f.releaseOnce.Do(func() { close(f.release) }) } +func (f *relayOwnerFixture) releaseClose() { f.closeReleaseOnce.Do(func() { close(f.closeRelease) }) } + +func (f *relayOwnerFixture) slots() (streams, connections, source int) { + if f.tlsServer != nil { + return len(f.tlsServer.core.sem), len(f.tlsServer.connSem), f.tlsServer.clients.count("127.0.0.1") + } + if f.quicServer != nil { + return len(f.quicServer.core.sem), len(f.quicServer.connSem), f.quicServer.clients.count("127.0.0.1") + } + return 0, 0, 0 +} + +func (f *relayOwnerFixture) assertSlots(streams int, parent bool) { + f.t.Helper() + s, connections, source := f.slots() + if s != streams || (parent && (connections != 1 || source != 1)) { + f.t.Errorf("stream/connection/source slots=%d/%d/%d, want stream=%d parent=%t", s, connections, source, streams, parent) + } + if f.quicServer != nil && len(f.quicServer.streamSem) != streams { + f.t.Errorf("native stream slots=%d, want %d", len(f.quicServer.streamSem), streams) + } +} + +func (f *relayOwnerFixture) halfClose() { + f.t.Helper() + writer, ok := f.conn.(interface{ CloseWrite() error }) + if !ok { + f.t.Fatalf("%T lacks real CloseWrite", f.conn) + } + if err := writer.CloseWrite(); err != nil { + f.t.Fatal(err) + } + f.require(f.originEOF, "actual target upload EOF") + f.mu.Lock() + destination := f.destination + f.mu.Unlock() + f.require(destination.halfClosed, "completed destination TCP CloseWrite") + // The upload pump has observed EOF and completed its only remaining I/O; + // no target response is emitted until independent releaseTarget. + f.assertSlots(1, true) +} + +func (f *relayOwnerFixture) stopOwner(action string) (<-chan struct{}, *error) { + done := make(chan struct{}) + var err error + f.register(done) + go func() { + defer close(done) + if action == "close" { + err = f.closeServer() + } else { + f.serveCancel() + <-f.serveDone + err = f.serveErr + } + }() + return done, &err +} + +func (f *relayOwnerFixture) assertStopped(done <-chan struct{}, result *error) bool { + if !f.wait(done, "owner shutdown (target independently still held)", 500*time.Millisecond) { + f.assertSlots(1, false) + return false + } + if *result != nil && !errors.Is(*result, net.ErrClosed) { + f.t.Errorf("owner shutdown: %v", *result) + } + f.assertSlots(0, false) + if s, connections, source := f.slots(); s != 0 || connections != 0 || source != 0 { + f.t.Errorf("completed owner retained slots %d/%d/%d", s, connections, source) + } + f.require(f.serveDone, "Serve after completed owner shutdown") + return true +} + +func (f *relayOwnerFixture) awaitRelayRelease() { + f.t.Helper() + timer, ticker := time.NewTimer(2*time.Second), time.NewTicker(10*time.Millisecond) + defer timer.Stop() + defer ticker.Stop() + for { + streams, _, _ := f.slots() + if streams == 0 && (f.quicServer == nil || len(f.quicServer.streamSem) == 0) { + f.assertSlots(0, false) + return + } + select { + case <-ticker.C: + case <-timer.C: + f.t.Error("completed healthy relay retained its stream lease") + return + } + } +} + +func (f *relayOwnerFixture) cleanup() { + f.releaseTarget() + f.releaseClose() + f.mu.Lock() + f.closing = true + origin, destination := f.origin, f.destination + f.mu.Unlock() + if origin != nil { + _ = origin.Close() + } + if f.listener != nil { + _ = f.listener.Close() + } + if destination != nil { + _ = destination.TCPConn.Close() + } + if f.conn != nil { + _ = f.conn.Close() + } + if f.closeClient != nil { + _ = f.closeClient() + } + if f.serveCancel != nil { + f.serveCancel() + } + if f.closeServer != nil { + done := make(chan struct{}) + f.register(done) + go func() { defer close(done); _ = f.closeServer() }() + } + f.mu.Lock() + workers := append([]<-chan struct{}(nil), f.workers...) + f.mu.Unlock() + for _, done := range workers { + f.wait(done, "independent cleanup worker join", 2*time.Second) + } + if f.closeServer != nil { + f.assertSlots(0, false) + if streams, connections, source := f.slots(); streams != 0 || connections != 0 || source != 0 { + f.t.Errorf("joined cleanup retained stream/connection/source slots %d/%d/%d", streams, connections, source) + } + } + if f.dials.Load() > 1 { + f.t.Errorf("unexpected destination dial count %d", f.dials.Load()) + } +} + +func TestRelayOwnerShutdownReleasesStalledDestination(t *testing.T) { + for _, mode := range []string{"tls", "quic"} { + for _, direction := range []string{"client_open", "client_halfclosed"} { + for _, action := range []string{"close", "serve_cancel"} { + t.Run(mode+"/"+direction+"/"+action, func(t *testing.T) { + f := newRelayOwnerFixture(t, mode, "stalled", false) + if direction == "client_halfclosed" { + f.halfClose() + } + done, result := f.stopOwner(action) + if f.assertStopped(done, result) { + f.require(f.originEOF, "actual target peer EOF before fixture cleanup") + f.mu.Lock() + destination := f.destination + f.mu.Unlock() + f.require(destination.closed, "real destination FD closed before shutdown completion") + } + // Negative controls reach independent cleanup without a Fatal or + // an origin deadline being counted as successful shutdown. + }) + } + } + } +} + +func TestRelayOwnerCancellationPreservesDelayedHalfCloseResponse(t *testing.T) { + for _, mode := range []string{"tls", "quic"} { + t.Run(mode, func(t *testing.T) { + f := newRelayOwnerFixture(t, mode, "delayed", false) + f.halfClose() + f.mu.Lock() + deadline := f.dialDeadline + f.mu.Unlock() + if deadline.IsZero() { + t.Fatal("actual destination dial lacked its configured deadline") + } + timer := time.NewTimer(max(0, time.Until(deadline))) + defer timer.Stop() + <-timer.C + if f.serveContext.Err() != nil { + t.Fatal("normal half-close canceled relay owner") + } + f.assertSlots(1, true) + f.releaseTarget() + response, err := io.ReadAll(f.conn) + if err != nil || !bytes.Equal(response, f.response) { + t.Errorf("delayed full response: %d/%d bytes, %v", len(response), len(f.response), err) + } + f.require(f.originDone, "delayed actual target response worker") + if f.originErr != nil { + t.Errorf("delayed target: %v", f.originErr) + } + f.awaitRelayRelease() + if f.serveContext.Err() != nil { + t.Error("successful delayed relay canceled parent") + } + }) + } +} + +func TestRelayOwnerShutdownJoinsDestinationCloseCallback(t *testing.T) { + for _, mode := range []string{"tls", "quic"} { + t.Run(mode, func(t *testing.T) { + f := newRelayOwnerFixture(t, mode, "stalled", true) + f.halfClose() + done, result := f.stopOwner("close") + f.mu.Lock() + destination := f.destination + f.mu.Unlock() + if !f.wait(destination.closeEntered, "owner triggered real target Close", 500*time.Millisecond) { + return + } + f.require(destination.closed, "real target FD closed before synthetic Close-return gate") + if destination.closeErr != nil { + t.Errorf("first actual target FD Close: %v", destination.closeErr) + } + timer := time.NewTimer(75 * time.Millisecond) + select { + case <-done: + t.Error("owner shutdown returned before target Close callback completed") + case <-timer.C: + } + timer.Stop() + f.assertSlots(1, false) + // This is a Close-completion gate, not a fake FD or a blocked-I/O + // witness. It verifies that release is ordered after owned cleanup. + f.releaseClose() + f.assertStopped(done, result) + if destination.closeCalls.Load() != 1 { + t.Errorf("actual target Close calls=%d, want 1", destination.closeCalls.Load()) + } + }) + } +} + +func TestRelayOwnerCancellationPreservesDestinationHalfClose(t *testing.T) { + for _, mode := range []string{"tls", "quic"} { + t.Run(mode, func(t *testing.T) { + f := newRelayOwnerFixture(t, mode, "response_first", false) + response, err := io.ReadAll(f.conn) + if err != nil || !bytes.Equal(response, f.response) { + t.Fatalf("actual response FIN/full bytes: %d/%d, %v", len(response), len(f.response), err) + } + if f.serveContext.Err() != nil { + t.Fatal("destination FIN canceled relay owner") + } + f.assertSlots(1, true) + if _, err := io.Copy(f.conn, bytes.NewReader(f.tail)); err != nil { + t.Errorf("upload after full response EOF: %v", err) + } + if err := f.conn.(interface{ CloseWrite() error }).CloseWrite(); err != nil { + t.Errorf("late upload FIN: %v", err) + } + f.require(f.originDone, "actual complete tail/EOF after destination FIN") + if f.originErr != nil { + t.Errorf("destination reverse half-close: %v", f.originErr) + } + f.require(f.originEOF, "target received complete tail and upload EOF") + f.awaitRelayRelease() + if f.serveContext.Err() != nil { + t.Error("normal duplex completion canceled parent") + } + }) + } +} diff --git a/internal/tunnel/tls.go b/internal/tunnel/tls.go index 151f75c..1e0fcb3 100644 --- a/internal/tunnel/tls.go +++ b/internal/tunnel/tls.go @@ -187,7 +187,9 @@ func (s *TLSServer) serveTLSConnection(ctx context.Context, conn *tls.Conn, sour if err != nil || conn.ConnectionState().NegotiatedProtocol != protocol.ALPN { return } - s.core.handleStream(ctx, conn, nil, nil) + s.core.handleStream(ctx, conn, nil, func(owner context.Context, stream deadlineConn, _ protocol.Request) streamRequestOptions { + return streamRequestOptions{Relay: stream, RelayOwner: owner} + }) } type sourceConnectionLimiter struct { From 97e71b88b7bedc0579655a7b525871ddaccec844 Mon Sep 17 00:00:00 2001 From: cppla Date: Sat, 3 Oct 2026 12:27:16 +0800 Subject: [PATCH 2/2] Retry only ephemeral UDP collisions in trailer test setup --- .../tunnel/web_cover_request_trailers_test.go | 2 +- internal/tunnel/web_ephemeral_fixture_test.go | 183 ++++++++++++++++++ 2 files changed, 184 insertions(+), 1 deletion(-) create mode 100644 internal/tunnel/web_ephemeral_fixture_test.go diff --git a/internal/tunnel/web_cover_request_trailers_test.go b/internal/tunnel/web_cover_request_trailers_test.go index 9905147..3f5b886 100644 --- a/internal/tunnel/web_cover_request_trailers_test.go +++ b/internal/tunnel/web_cover_request_trailers_test.go @@ -211,7 +211,7 @@ func newWebCoverRequestTrailersFixture(t *testing.T, constructor string, proto i website.ServeHTTP(w, r) }) serverTLS, clientTLS := testTLSConfigs(t) - f.server, err = ListenWeb(WebServerConfig{TCPAddress: "127.0.0.1:0", UDPAddress: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, Cover: observingCover, + f.server, err = listenWebEphemeralFixture(WebServerConfig{TCPAddress: "127.0.0.1:0", UDPAddress: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, Cover: observingCover, Dialer: transport.DialFunc(func(context.Context, string, string) (net.Conn, error) { f.tunnelDials.Add(1) return nil, errors.New("request-trailer fixture forbids tunnel dial") diff --git a/internal/tunnel/web_ephemeral_fixture_test.go b/internal/tunnel/web_ephemeral_fixture_test.go new file mode 100644 index 0000000..6e5d76a --- /dev/null +++ b/internal/tunnel/web_ephemeral_fixture_test.go @@ -0,0 +1,183 @@ +package tunnel + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "runtime" + "syscall" + "testing" + + "github.com/cppla/autocar/internal/transport" +) + +const webEphemeralFixtureAttempts = 32 + +// Test setup only: TCP chooses a free ephemeral port before ListenWeb binds +// UDP to that same port. The independent UDP namespace may already own it. +// Never use this helper for fixed-address policy or rollback assertions. +func listenWebEphemeralFixture(config WebServerConfig) (*WebServer, error) { + return listenWebEphemeralFixtureWith(config, ListenWeb) +} + +func listenWebEphemeralFixtureWith(config WebServerConfig, listen func(WebServerConfig) (*WebServer, error)) (*WebServer, error) { + if config.TCPAddress != "127.0.0.1:0" || config.UDPAddress != "127.0.0.1:0" { + return listen(config) + } + for attempt := 0; ; attempt++ { + server, err := listen(config) + if server != nil || err == nil || attempt+1 == webEphemeralFixtureAttempts || !webEphemeralFixtureUDPCollision(err) { + return server, err + } + } +} + +func webEphemeralFixtureUDPCollision(err error) bool { + // A rollback failure joined to the bind error must not be hidden by setup + // retry. Accept a single wrapping chain only, not any matching joined leaf. + for current := err; current != nil; current = errors.Unwrap(current) { + if _, joined := current.(interface{ Unwrap() []error }); joined { + return false + } + } + var opErr *net.OpError + if !errors.As(err, &opErr) || opErr.Op != "listen" || opErr.Net != "udp" { + return false + } + address, ok := opErr.Addr.(*net.UDPAddr) + return ok && address != nil && address.Zone == "" && address.Port > 0 && address.Port <= 65535 && + address.IP.Equal(net.IPv4(127, 0, 0, 1)) && isWebTestAddressInUse(opErr.Err) +} + +func TestWebEphemeralFixtureListenRetriesRealUDPPortCollision(t *testing.T) { + serverTLS, _ := testTLSConfigs(t) + config := WebServerConfig{ + TCPAddress: "127.0.0.1:0", UDPAddress: "127.0.0.1:0", Token: testToken, + TLSConfig: serverTLS, Cover: http.NotFoundHandler(), + Dialer: transport.DialFunc(func(context.Context, string, string) (net.Conn, error) { + return nil, errors.New("ephemeral listener fixture forbids destination dialing") + }), + } + reservedTCP, occupiedUDP := reserveWebTestTCPUDP(t) + occupiedAddress := occupiedUDP.LocalAddr().String() + if err := reservedTCP.Close(); err != nil { + t.Fatal(err) + } + var firstError error + calls := 0 + server, err := listenWebEphemeralFixtureWith(config, func(attempt WebServerConfig) (*WebServer, error) { + calls++ + if attempt.TCPAddress != config.TCPAddress || attempt.UDPAddress != config.UDPAddress { + t.Fatalf("helper changed caller's ephemeral addresses: %q/%q", attempt.TCPAddress, attempt.UDPAddress) + } + if calls > 1 { + // All retries go through the real normal :0 constructor. A later + // unrelated ephemeral collision is still subject to the same bound. + return ListenWeb(attempt) + } + // The controlled first call uses an actually occupied owned UDP port, + // not a fabricated bind error. TCP is available in its own namespace. + attempt.TCPAddress, attempt.UDPAddress = occupiedAddress, occupiedAddress + unexpected, err := ListenWeb(attempt) + if unexpected != nil { + _ = unexpected.Close() + t.Fatal("actual occupied UDP constructor unexpectedly returned a server") + } + requireWebListenAddressInUse(t, err, "udp", occupiedAddress) + firstError = err + // Prove rollback before any retry: no retrying bind or helper is used + // for this assertion, so a leaked TCP listener cannot be concealed. + probe, probeErr := net.Listen("tcp", occupiedAddress) + if probeErr != nil { + t.Fatalf("original ListenWeb failed to roll back TCP: %v", probeErr) + } + if closeErr := probe.Close(); closeErr != nil { + t.Fatal(closeErr) + } + return nil, err + }) + if server != nil { + t.Cleanup(func() { + if err := normalizeWebServerCloseError(server.Close()); err != nil { + t.Errorf("owned ephemeral server cleanup: %v", err) + } + }) + } + if err != nil || server == nil { + t.Fatalf("normal ephemeral setup after actual UDP collision: server=%t calls=%d err=%v", server != nil, calls, err) + } + if firstError == nil || calls < 2 || calls > webEphemeralFixtureAttempts { + t.Fatalf("real collision/retry evidence: first=%v calls=%d", firstError, calls) + } + if server.TCPAddr().String() != server.UDPAddr().String() { + t.Fatalf("successful retry changed shared TCP/UDP authority: %s/%s", server.TCPAddr(), server.UDPAddr()) + } + if config.TCPAddress != "127.0.0.1:0" || config.UDPAddress != "127.0.0.1:0" { + t.Fatal("fixture mutated the caller's config") + } + t.Logf("actual owned UDP bind failure, immediate TCP rollback, and normal shared-port retry succeeded (%d attempts)", calls) +} + +func TestWebEphemeralFixtureListenRetryPolicy(t *testing.T) { + // Pure policy controls below are synthetic, not network observations. The + // top above separately proves real ListenWeb bind failure and TCP rollback. + nativeErrno := syscall.EADDRINUSE + if runtime.GOOS == "windows" { + nativeErrno = syscall.Errno(10048) + } + address := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 43217} + collision := &net.OpError{Op: "listen", Net: "udp", Addr: address, Err: nativeErrno} + ordinary := errors.New("private ordinary setup error") + tests := []struct { + name string + tcp, udp string + err error + calls int + }{ + {"udp_collision_bound", "127.0.0.1:0", "127.0.0.1:0", collision, webEphemeralFixtureAttempts}, + {"wrapped_udp_collision_bound", "127.0.0.1:0", "127.0.0.1:0", fmt.Errorf("constructor: %w", collision), webEphemeralFixtureAttempts}, + {"ordinary_error", "127.0.0.1:0", "127.0.0.1:0", ordinary, 1}, + {"matching_text_only", "127.0.0.1:0", "127.0.0.1:0", errors.New("listen udp 127.0.0.1:43217: address already in use"), 1}, + {"bare_errno", "127.0.0.1:0", "127.0.0.1:0", nativeErrno, 1}, + {"tcp_collision", "127.0.0.1:0", "127.0.0.1:0", &net.OpError{Op: "listen", Net: "tcp", Addr: address, Err: nativeErrno}, 1}, + {"udp4_not_production_network", "127.0.0.1:0", "127.0.0.1:0", &net.OpError{Op: "listen", Net: "udp4", Addr: address, Err: nativeErrno}, 1}, + {"non_listen_operation", "127.0.0.1:0", "127.0.0.1:0", &net.OpError{Op: "dial", Net: "udp", Addr: address, Err: nativeErrno}, 1}, + {"permission_error", "127.0.0.1:0", "127.0.0.1:0", &net.OpError{Op: "listen", Net: "udp", Addr: address, Err: syscall.EACCES}, 1}, + {"non_loopback_error_address", "127.0.0.1:0", "127.0.0.1:0", &net.OpError{Op: "listen", Net: "udp", Addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 1), Port: 43217}, Err: nativeErrno}, 1}, + {"missing_error_address", "127.0.0.1:0", "127.0.0.1:0", &net.OpError{Op: "listen", Net: "udp", Err: nativeErrno}, 1}, + {"joined_rollback_error", "127.0.0.1:0", "127.0.0.1:0", errors.Join(collision, ordinary), 1}, + {"joined_inner_error", "127.0.0.1:0", "127.0.0.1:0", &net.OpError{Op: "listen", Net: "udp", Addr: address, Err: errors.Join(nativeErrno, ordinary)}, 1}, + {"fixed_tcp", "127.0.0.1:43217", "127.0.0.1:0", collision, 1}, + {"fixed_udp", "127.0.0.1:0", "127.0.0.1:43217", collision, 1}, + {"both_fixed", "127.0.0.1:43217", "127.0.0.1:43217", collision, 1}, + {"inherited_udp", "127.0.0.1:0", "", collision, 1}, + {"wildcard_tcp", ":0", "127.0.0.1:0", collision, 1}, + {"dns_authority", "localhost:0", "127.0.0.1:0", collision, 1}, + {"ipv6_ephemeral", "[::1]:0", "[::1]:0", collision, 1}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + calls := 0 + server, err := listenWebEphemeralFixtureWith(WebServerConfig{TCPAddress: test.tcp, UDPAddress: test.udp}, func(WebServerConfig) (*WebServer, error) { + calls++ + return nil, test.err + }) + if server != nil || err != test.err || calls != test.calls { + t.Fatalf("error identity/bounded attempts: server=%t exact=%t calls=%d; want %d", server != nil, err == test.err, calls, test.calls) + } + }) + } + t.Run("successful_result_not_retried", func(t *testing.T) { + owned := &WebServer{} + calls := 0 + server, err := listenWebEphemeralFixtureWith(WebServerConfig{TCPAddress: "127.0.0.1:0", UDPAddress: "127.0.0.1:0"}, func(WebServerConfig) (*WebServer, error) { + calls++ + return owned, nil + }) + if server != owned || err != nil || calls != 1 { + t.Fatalf("successful constructor changed: exact=%t err=%v calls=%d", server == owned, err, calls) + } + }) +}