diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index ff00ab5..36e986c 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -264,6 +264,15 @@ shutdown context error and initiates request cancellation and socket cleanup; it does not wait indefinitely for custom handlers that ignore cancellation. This change is not included in v1.0.1. +Source builds also cancel pending SOCKS5 TCP CONNECT setup when its tracked +client connection is forcibly closed, without waiting for the full dial +timeout. Successful setup still cancels the temporary dialing context without +tying the established tunnel to that timeout. There is no competing TCP reader: +early payload and write-half-close remain usable. Cancellation affects only +that caller's wait, not a shared native client's physical dial. This does not +add immediate peer-disconnect detection during a pending TCP dial, and is not +included in v1.0.1. + For SOCKS5 TCP and HTTP CONNECT tunnels, `--idle-timeout` (default `5m`) measures inactivity across both directions: an active download or upload does not need reverse-direction application traffic to stay open. A blocked write diff --git a/internal/proxy/lifecycle.go b/internal/proxy/lifecycle.go index d0cffad..82db4f6 100644 --- a/internal/proxy/lifecycle.go +++ b/internal/proxy/lifecycle.go @@ -99,10 +99,11 @@ type trackedConn struct { connectionCancel context.CancelFunc } -// connectionContext binds net/http's connection context to explicit socket -// closure. Its background reader can stop after caching a pipelined byte, so -// closing the socket alone need not produce a read error that cancels a request. -// ConnContext calls this once per accepted connection. Registration and Close +// connectionContext binds frontend work to explicit socket closure. A caller +// waiting for upstream setup need not be reading the socket; net/http's reader +// can also stop after caching a pipelined byte. Socket closure alone therefore +// need not cancel that work. The frontend calls this once per accepted +// connection, from HTTP ConnContext or SOCKS TCP setup. Registration and Close // may race during shutdown; either ordering must return a canceled context. func (c *trackedConn) connectionContext(parent context.Context) context.Context { ctx, cancel := context.WithCancel(parent) diff --git a/internal/proxy/socks5.go b/internal/proxy/socks5.go index ee9ce64..17b2348 100644 --- a/internal/proxy/socks5.go +++ b/internal/proxy/socks5.go @@ -137,11 +137,23 @@ func (s *SOCKS5Server) serveConn(client net.Conn) { func (s *SOCKS5Server) serveConnect(client net.Conn, request socksRequest) { ctx := context.Background() - cancel := func() {} + if tracked, ok := client.(*trackedConn); ok { + // Bind only this caller's setup wait to full socket close. Do not add + // a reader that could consume early payload or treat half-close as + // abandonment, and do not close the shared upstream dialer. + ctx = tracked.connectionContext(ctx) + } + var cancel context.CancelFunc if s.cfg.dialTimeout > 0 { ctx, cancel = context.WithTimeout(ctx, s.cfg.dialTimeout) + } else { + ctx, cancel = context.WithCancel(ctx) + } + var upstream net.Conn + err := ctx.Err() + if err == nil { + upstream, err = s.cfg.dialer.DialContext(ctx, "tcp", request.address) } - upstream, err := s.cfg.dialer.DialContext(ctx, "tcp", request.address) cancel() if s.cfg.handshakeTimeout > 0 { _ = client.SetWriteDeadline(time.Now().Add(s.cfg.handshakeTimeout)) diff --git a/internal/proxy/socks5_tcp_setup_cancel_test.go b/internal/proxy/socks5_tcp_setup_cancel_test.go new file mode 100644 index 0000000..7c6cf5f --- /dev/null +++ b/internal/proxy/socks5_tcp_setup_cancel_test.go @@ -0,0 +1,357 @@ +package proxy + +import ( + "bytes" + "context" + "crypto/tls" + "errors" + "io" + "net" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" + "github.com/cppla/autocar/internal/tunnel" +) + +// Expose completion of one production serveConn worker without changing its +// protocol, reads, deadlines or context. Both ends use real loopback TCP and +// the accepted connection still passes through the ordinary managed listener. +type socksTCPForcedSetupFixture struct { + server *SOCKS5Server + client net.Conn + handlerDone chan struct{} +} + +func newSOCKSTCPForcedSetupFixture(t *testing.T, cfg Config) *socksTCPForcedSetupFixture { + t.Helper() + server, err := NewSOCKS5Server(cfg) + if err != nil { + t.Fatal(err) + } + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + managed, err := server.lifecycle.manage(listener) + if err != nil { + _ = listener.Close() + t.Fatal(err) + } + f := &socksTCPForcedSetupFixture{server: server, handlerDone: make(chan struct{})} + acceptResult := make(chan error, 1) + go func() { + defer close(f.handlerDone) + conn, err := managed.Accept() + acceptResult <- err + if err == nil { + server.serveConn(conn) + } + }() + t.Cleanup(func() { + if f.client != nil { + _ = f.client.Close() + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _ = server.Shutdown(ctx) + awaitSOCKSTCPForcedSignal(t, f.handlerDone, "serveConn cleanup") + }) + f.client = dialTCP(t, listener.Addr().String()) + select { + case err := <-acceptResult: + if err != nil { + t.Fatal(err) + } + case <-time.After(2 * time.Second): + t.Fatal("SOCKS TCP fixture did not accept its client") + } + socksGreeting(t, f.client, nil) + mustWrite(t, f.client, ipv4SOCKSRequest(socksCommandConnect, net.IPv4(127, 0, 0, 1), 443)) + return f +} + +func awaitSOCKSTCPForcedSignal(t *testing.T, done <-chan struct{}, label string) { + t.Helper() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Errorf("%s did not finish", label) + } +} + +func awaitSOCKSTCPForcedContext(t *testing.T, contexts <-chan context.Context) context.Context { + t.Helper() + select { + case ctx := <-contexts: + return ctx + case <-time.After(2 * time.Second): + t.Fatal("SOCKS TCP did not enter the caller DialContext") + return nil + } +} + +func forceSOCKSTCPSetupShutdown(t *testing.T, f *socksTCPForcedSetupFixture) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + err := f.server.Shutdown(ctx) + cancel() + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("forced Shutdown = %v, want deadline exceeded", err) + } + f.server.lifecycle.tracker.mu.Lock() + held := len(f.server.lifecycle.tracker.conns) + f.server.lifecycle.tracker.mu.Unlock() + if held != 0 { + t.Fatalf("forced Shutdown retained %d tracked connection slots", held) + } + _ = f.client.SetReadDeadline(time.Now().Add(time.Second)) + if _, err := io.ReadAll(f.client); err != nil { + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + t.Fatalf("forced Shutdown did not close the downstream TCP connection: %v", err) + } + } +} + +func requireSOCKSTCPForcedSetupCompletion(t *testing.T, ctx context.Context, dialDone, handlerDone <-chan struct{}) { + t.Helper() + // One budget covers the actual caller context, dial return and production + // handler completion. It is much shorter than the native default 5s dial. + timer := time.NewTimer(time.Second) + defer timer.Stop() + ctxDone := ctx.Done() + for ctxDone != nil || dialDone != nil || handlerDone != nil { + select { + case <-ctxDone: + ctxDone = nil + case <-dialDone: + dialDone = nil + case <-handlerDone: + handlerDone = nil + case <-timer.C: + t.Fatalf("forced setup remained active: context=%v dialPending=%t handlerPending=%t", ctx.Err(), dialDone != nil, handlerDone != nil) + } + } + if !errors.Is(ctx.Err(), context.Canceled) { + t.Fatalf("forced setup caller error = %v, want context canceled", ctx.Err()) + } +} + +func newSOCKSTCPNativeBlackhole(t *testing.T) (*tunnel.Client, *net.UDPConn) { + t.Helper() + blackhole, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = blackhole.Close() }) + native, err := tunnel.NewClient(tunnel.ClientConfig{ + ServerAddress: blackhole.LocalAddr().String(), + Token: "local-socks-tcp-setup-regression-token", + TLSConfig: &tls.Config{ServerName: "localhost"}, + // Preserve the production default 5s shared physical-dial budget. + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = native.Close() }) + return native, blackhole +} + +func readSOCKSTCPNativeInitial(t *testing.T, peer *net.UDPConn) (*net.UDPAddr, []byte) { + t.Helper() + if err := peer.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil { + t.Fatal(err) + } + packet := make([]byte, 4096) + n, source, err := peer.ReadFromUDP(packet) + if err != nil || n < 1200 || packet[0]&0xc0 != 0xc0 { + t.Fatalf("native QUIC Initial was not observed: bytes=%d error=%v", n, err) + } + // The public long-header destination CID identifies this particular + // physical handshake without reaching into native Client internals. A + // single-use client may legitimately use an empty source CID. + destinationLength := int(packet[5]) + if destinationLength == 0 || 6+destinationLength >= n { + t.Fatal("native Initial has no complete destination connection ID") + } + destinationID := bytes.Clone(packet[6 : 6+destinationLength]) + t.Logf("real native QUIC Initial: bytes=%d source=%s; shared dial budget remains 5s", n, source) + return source, destinationID +} + +func TestSOCKSTCPSetupAlreadyClosedSkipsDial(t *testing.T) { + dialCalls := 0 + dialer := transport.DialFunc(func(context.Context, string, string) (net.Conn, error) { + dialCalls++ + return nil, errors.New("closed SOCKS connection must not start a remote dial") + }) + server, err := NewSOCKS5Server(Config{Dialer: dialer}) + if err != nil { + t.Fatal(err) + } + local, remote := net.Pipe() + t.Cleanup(func() { _ = remote.Close() }) + tracked := &trackedConn{Conn: local, tracker: server.lifecycle.tracker} + if !server.lifecycle.tracker.add(tracked) { + t.Fatal("fixture connection was not admitted") + } + if err := tracked.Close(); err != nil { + t.Fatal(err) + } + server.serveConnect(tracked, socksRequest{command: socksCommandConnect, address: "127.0.0.1:443"}) + if dialCalls != 0 { + t.Fatalf("already closed SOCKS TCP setup started %d remote dials", dialCalls) + } +} + +func TestSOCKSTCPSetupForcedShutdownCancelsDial(t *testing.T) { + for _, mode := range []string{"context-aware pending dial", "real native QUIC blackhole"} { + t.Run(mode, func(t *testing.T) { + contexts := make(chan context.Context, 1) + dialDone := make(chan struct{}) + release := make(chan struct{}) + var native *tunnel.Client + var peer *net.UDPConn + if mode == "real native QUIC blackhole" { + native, peer = newSOCKSTCPNativeBlackhole(t) + } + dialer := transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + defer close(dialDone) + contexts <- ctx + if native != nil { + return native.DialContext(ctx, network, address) + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-release: + return nil, errors.New("independent test cleanup release") + } + }) + f := newSOCKSTCPForcedSetupFixture(t, Config{Dialer: dialer, MaxConnections: 1}) + t.Cleanup(func() { + // These are independent failure-cleanup paths, used only after + // the assertions about frontend cancellation have completed. + close(release) + if native != nil { + _ = native.Close() + } + awaitSOCKSTCPForcedSignal(t, dialDone, "caller dial cleanup") + }) + ctx := awaitSOCKSTCPForcedContext(t, contexts) + if peer != nil { + readSOCKSTCPNativeInitial(t, peer) + } + if err := ctx.Err(); err != nil { + t.Fatalf("setup context expired before Shutdown: %v", err) + } + forceSOCKSTCPSetupShutdown(t, f) + requireSOCKSTCPForcedSetupCompletion(t, ctx, dialDone, f.handlerDone) + }) + } +} + +func TestSOCKSTCPSetupShutdownPreservesSharedNativeDial(t *testing.T) { + native, peer := newSOCKSTCPNativeBlackhole(t) + contexts := make(chan context.Context, 1) + firstDialDone := make(chan struct{}) + dialer := transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + defer close(firstDialDone) + contexts <- ctx + return native.DialContext(ctx, network, address) + }) + f := newSOCKSTCPForcedSetupFixture(t, Config{Dialer: dialer, MaxConnections: 1}) + firstCtx := awaitSOCKSTCPForcedContext(t, contexts) + originalSource, originalDestinationID := readSOCKSTCPNativeInitial(t, peer) + siblingCtx, cancelSibling := context.WithCancel(context.Background()) + siblingStarted, siblingDone := make(chan struct{}), make(chan struct{}) + siblingResult := make(chan error, 1) + go func() { + defer close(siblingDone) + close(siblingStarted) + conn, err := native.DialContext(siblingCtx, "tcp", "127.0.0.1:8443") + if conn != nil { + _ = conn.Close() + } + siblingResult <- err + }() + t.Cleanup(func() { + cancelSibling() + _ = native.Close() + awaitSOCKSTCPForcedSignal(t, siblingDone, "independent native caller cleanup") + awaitSOCKSTCPForcedSignal(t, firstDialDone, "frontend native caller cleanup") + }) + awaitSOCKSTCPForcedSignal(t, siblingStarted, "independent native caller start") + forceSOCKSTCPSetupShutdown(t, f) + requireSOCKSTCPForcedSetupCompletion(t, firstCtx, firstDialDone, f.handlerDone) + if err := siblingCtx.Err(); err != nil { + t.Fatalf("frontend Shutdown canceled the independent caller context: %v", err) + } + select { + case err := <-siblingResult: + t.Fatalf("frontend Shutdown ended the independent native caller: %v", err) + default: + } + + // Drain any packets already queued when the first frontend finished. A + // subsequently received packet from the same UDP source and destination CID + // proves the existing physical handshake continued rather than restarting. + if err := peer.SetReadDeadline(time.Now().Add(10 * time.Millisecond)); err != nil { + t.Fatal(err) + } + packet := make([]byte, 4096) + for { + _, _, err := peer.ReadFromUDP(packet) + if err == nil { + continue + } + var netErr net.Error + if !errors.As(err, &netErr) || !netErr.Timeout() { + t.Fatal(err) + } + break + } + if err := peer.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil { + t.Fatal(err) + } + n, nextSource, err := peer.ReadFromUDP(packet) + if err != nil { + t.Fatalf("shared physical QUIC dial stopped retransmitting after frontend Shutdown: %v", err) + } + if nextSource.String() != originalSource.String() { + t.Fatalf("shared physical dial changed its UDP source: %s -> %s", originalSource, nextSource) + } + if n < 1200 || packet[0]&0xc0 != 0xc0 { + t.Fatalf("shared dial retransmission is not a QUIC Initial: bytes=%d", n) + } + destinationLength := int(packet[5]) + if destinationLength == 0 || 6+destinationLength >= n { + t.Fatal("shared Initial retransmission has an invalid connection-ID encoding") + } + nextDestinationID := packet[6 : 6+destinationLength] + if !bytes.Equal(nextDestinationID, originalDestinationID) { + t.Fatalf("shared physical dial restarted with a different destination connection ID: %x -> %x", originalDestinationID, nextDestinationID) + } + if err := siblingCtx.Err(); err != nil { + t.Fatalf("independent caller context expired during the shared handshake: %v", err) + } + select { + case err := <-siblingResult: + t.Fatalf("independent caller stopped during the preserved shared handshake: %v", err) + default: + } + t.Logf("after frontend cancellation, original QUIC handshake retransmitted %d bytes from %s with the same destination CID", n, nextSource) + // Only now cancel this independent caller and verify its own wait returns, + // before fixture cleanup closes the separately owned shared native Client. + cancelSibling() + select { + case err := <-siblingResult: + if !errors.Is(err, context.Canceled) { + t.Fatalf("independent native caller cancellation = %v, want context canceled", err) + } + case <-time.After(time.Second): + t.Fatal("independent native caller did not cancel promptly") + } + awaitSOCKSTCPForcedSignal(t, siblingDone, "independently canceled native caller") +} diff --git a/internal/proxy/socks5_tcp_setup_preserve_test.go b/internal/proxy/socks5_tcp_setup_preserve_test.go new file mode 100644 index 0000000..6e1b88f --- /dev/null +++ b/internal/proxy/socks5_tcp_setup_preserve_test.go @@ -0,0 +1,289 @@ +package proxy + +import ( + "bytes" + "context" + "errors" + "io" + "net" + "sync" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +type socksTCPPreservePayload struct { + body []byte + err error +} + +// The fixture accepts through the production managedListener, then runs the +// unmodified serveConn with a completion signal. The dial gate delays a real +// TCP origin connection without reading any downstream application bytes. +type socksTCPPreserveFixture struct { + server *SOCKS5Server + client *net.TCPConn + originAddress string + setupContext chan context.Context + dialDone chan struct{} + dialRelease chan struct{} + dialOnce sync.Once + originPayload chan socksTCPPreservePayload + replyRelease chan struct{} + replyOnce sync.Once + originDone chan struct{} + originError error + serveDone chan struct{} + serveError error + mu sync.Mutex + closed bool + connections []net.Conn +} + +func (f *socksTCPPreserveFixture) releaseDial() { + f.dialOnce.Do(func() { close(f.dialRelease) }) +} + +func (f *socksTCPPreserveFixture) releaseReply() { + f.replyOnce.Do(func() { close(f.replyRelease) }) +} + +func (f *socksTCPPreserveFixture) ownConnection(conn net.Conn) bool { + f.mu.Lock() + if f.closed { + f.mu.Unlock() + _ = conn.Close() + return false + } + f.connections = append(f.connections, conn) + f.mu.Unlock() + return true +} + +func socksTCPPreserveWait(t *testing.T, done <-chan struct{}, label string) bool { + t.Helper() + select { + case <-done: + return true + case <-time.After(2 * time.Second): + t.Errorf("%s did not finish", label) + return false + } +} + +func newSOCKSTCPPreserveFixture(t *testing.T, response []byte) *socksTCPPreserveFixture { + t.Helper() + originListener, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + f := &socksTCPPreserveFixture{ + originAddress: originListener.Addr().String(), + setupContext: make(chan context.Context, 1), + dialDone: make(chan struct{}), + dialRelease: make(chan struct{}), + originPayload: make(chan socksTCPPreservePayload, 1), + replyRelease: make(chan struct{}), + originDone: make(chan struct{}), + serveDone: make(chan struct{}), + } + var proxyListener *net.TCPListener + workersStarted := false + t.Cleanup(func() { + // These releases work independently of production context propagation. + f.releaseDial() + f.releaseReply() + _ = originListener.Close() + if proxyListener != nil { + _ = proxyListener.Close() + } + f.mu.Lock() + f.closed = true + connections := append([]net.Conn(nil), f.connections...) + f.mu.Unlock() + for _, conn := range connections { + _ = conn.Close() + } + if f.client != nil { + _ = f.client.Close() + } + if f.server != nil { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _ = f.server.Shutdown(ctx) + } + if workersStarted { + socksTCPPreserveWait(t, f.serveDone, "SOCKS serveConn cleanup") + socksTCPPreserveWait(t, f.originDone, "TCP origin cleanup") + } + }) + + dialer := transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + defer close(f.dialDone) + f.setupContext <- ctx + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-f.dialRelease: + } + conn, err := (&net.Dialer{}).DialContext(ctx, network, address) + if err != nil { + return nil, err + } + if !f.ownConnection(conn) { + return nil, net.ErrClosed + } + return conn, nil + }) + f.server, err = NewSOCKS5Server(Config{ + Dialer: dialer, + DialTimeout: 750 * time.Millisecond, + HandshakeTimeout: 2 * time.Second, + IdleTimeout: 3 * time.Second, + MaxConnections: 1, + }) + if err != nil { + t.Fatal(err) + } + proxyListener, err = net.ListenTCP("tcp", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + managed, err := f.server.lifecycle.manage(proxyListener) + if err != nil { + t.Fatal(err) + } + workersStarted = true + go func() { + defer close(f.serveDone) + conn, err := managed.Accept() + if err != nil { + f.serveError = err + return + } + f.server.serveConn(conn) + }() + go func() { + defer close(f.originDone) + conn, err := originListener.AcceptTCP() + if err != nil { + f.originError = err + return + } + defer conn.Close() + if !f.ownConnection(conn) { + f.originError = net.ErrClosed + return + } + if err := conn.SetDeadline(time.Now().Add(5 * time.Second)); err != nil { + f.originError = err + return + } + body, err := io.ReadAll(conn) + f.originPayload <- socksTCPPreservePayload{body: body, err: err} + if err != nil { + f.originError = err + return + } + <-f.replyRelease + if _, err := io.Copy(conn, bytes.NewReader(response)); err != nil { + f.originError = err + return + } + f.originError = conn.CloseWrite() + }() + f.client, err = net.DialTCP("tcp", nil, proxyListener.Addr().(*net.TCPAddr)) + if err != nil { + t.Fatal(err) + } + if err := f.client.SetDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatal(err) + } + return f +} + +func TestSOCKS5TCPSetupPreservesEarlyBytesAndHalfClose(t *testing.T) { + for _, crossSetupDeadline := range []bool{false, true} { + name := "reply-after-setup" + if crossSetupDeadline { + name = "reply-after-original-setup-deadline" + } + t.Run(name, func(t *testing.T) { + payload := bytes.Repeat([]byte("\x00\x16\x03\x01 early SOCKS TCP payload\xff"), 1024) + response := bytes.Repeat([]byte("reverse reply after upload EOF\x00"), 256) + f := newSOCKSTCPPreserveFixture(t, response) + socksGreeting(t, f.client, nil) + host, port := splitAddress(t, f.originAddress) + mustWrite(t, f.client, domainSOCKSRequest(socksCommandConnect, host, port)) + var setupCtx context.Context + select { + case setupCtx = <-f.setupContext: + case <-time.After(2 * time.Second): + t.Fatal("SOCKS CONNECT did not enter pending DialContext") + } + setupDeadline, ok := setupCtx.Deadline() + if !ok { + t.Fatal("configured SOCKS setup context has no deadline") + } + // Only the client writes these bytes. There is no extra downstream + // reader while DialContext is pending, even after the write half-close. + mustWrite(t, f.client, payload) + if err := f.client.CloseWrite(); err != nil { + t.Fatal(err) + } + select { + case <-setupCtx.Done(): + t.Fatalf("early data/write half-close canceled pending setup: %v", setupCtx.Err()) + case <-time.After(30 * time.Millisecond): + } + f.releaseDial() + if reply := readSOCKSReply(t, f.client); reply != socksReplySucceeded { + t.Fatalf("CONNECT reply = %d, want success", reply) + } + if !socksTCPPreserveWait(t, f.dialDone, "completed TCP dial") { + return + } + if !errors.Is(setupCtx.Err(), context.Canceled) { + t.Fatalf("successful setup retained its context: %v", setupCtx.Err()) + } + select { + case received := <-f.originPayload: + if received.err != nil || !bytes.Equal(received.body, payload) { + t.Fatalf("origin payload = %d/%d bytes, equal=%t, error=%v", len(received.body), len(payload), bytes.Equal(received.body, payload), received.err) + } + case <-time.After(2 * time.Second): + t.Fatal("origin did not receive intact early payload followed by EOF") + } + if crossSetupDeadline { + // The setup context is already canceled, so wait on the saved + // original deadline rather than on that context's Done channel. + if delay := time.Until(setupDeadline.Add(50 * time.Millisecond)); delay > 0 { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-timer.C: + case <-f.serveDone: + t.Fatal("relay ended before reply across original setup deadline") + } + } + } + f.releaseReply() + got, err := io.ReadAll(f.client) + if err != nil || !bytes.Equal(got, response) { + t.Fatalf("reverse reply = %d/%d bytes, equal=%t, error=%v", len(got), len(response), bytes.Equal(got, response), err) + } + if socksTCPPreserveWait(t, f.originDone, "completed TCP origin") && f.originError != nil { + t.Errorf("origin: %v", f.originError) + } + if socksTCPPreserveWait(t, f.serveDone, "completed SOCKS serveConn") && f.serveError != nil { + t.Errorf("serveConn fixture: %v", f.serveError) + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := f.server.lifecycle.tracker.wait(ctx); err != nil { + t.Errorf("finished healthy relay retained its accepted slot: %v", err) + } + }) + } +}