diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index a308468..aafc0ef 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -47,6 +47,17 @@ dependency-boundary check permits only the exact fork version above, rejects local replacement or copied/vendored source trees, and scans tracked and untracked Go source for the known disallowed proxy application-module import. +## Local TCP frontend lifecycle + +HTTP CONNECT and SOCKS5 CONNECT keep the accepted connection's lifetime +separate from the destination dial timeout. After setup succeeds, ordinary +upload EOF only half-closes the destination writer and permits a delayed +response. Explicit socket closure, including forced frontend shutdown, closes +the owned destination even when the upload pump has already finished. The +relay joins both pumps and any started cancellation close callback before +returning; it never closes the shared transport dialer. HTTP CONNECT installs +this owner before replaying any bytes buffered during request parsing. + ## Native TCP flow 1. The local frontend authenticates the local user when configured and parses @@ -107,6 +118,11 @@ connection and wake waiters to select a replacement. Reconnects and the new H2 connection selected after GOAWAY run the full bootstrap again. QUIC migration or NAT rebinding that remains the same `*quic.Conn` retains authentication state. +Stopping an opening cancellation watcher is not the connection handoff. H2 +and H3 recheck the caller's cancellation and deadline under their final +ownership lock before publishing a TCP stream. Canceling an already handed-off +caller context does not close the healthy stream or its shared connection. + The cover is either a local static directory or a reverse proxy to one fixed, operator-authorized HTTP(S) origin. Ordinary combined H1/H2/H3 cover responses advertise the bound H3 service through `Alt-Svc`; raw upgraded 101 responses do @@ -181,6 +197,11 @@ logical PacketConn, but the H3 implementation creates one authenticated request stream lazily for each normalized target and reuses that stream for later datagrams to the same target. +A successful target open that races with PacketConn closure remains a pending +owner until its client stream, physical-session reservation and UDP admission +slot have been released. Only then is opening completion published to Close +and other waiters; cleanup runs outside the packet mutex. + The SOCKS frontend also isolates lazy H3 target opening, but only for transports that explicitly implement `transport.PacketConcurrentSender`. Canonical target aliases share one FIFO lane; the WebClient and close-once wrappers preserve diff --git a/internal/proxy/http.go b/internal/proxy/http.go index 54a40a1..baaeb10 100644 --- a/internal/proxy/http.go +++ b/internal/proxy/http.go @@ -183,6 +183,9 @@ func (s *HTTPServer) serveConnect(w http.ResponseWriter, r *http.Request) { return } defer client.Close() + // Hijacked response writes and buffered-prefix replay are already tunnel + // work: bind their lifetime before either can block, not only the relay. + defer watchRelayOwner(r.Context(), client, upstream)() if _, err := buffered.WriteString("HTTP/1.1 200 Connection Established\r\n\r\n"); err != nil { return diff --git a/internal/proxy/relay.go b/internal/proxy/relay.go index 7bfaf3c..9413c88 100644 --- a/internal/proxy/relay.go +++ b/internal/proxy/relay.go @@ -1,6 +1,7 @@ package proxy import ( + "context" "errors" "io" "net" @@ -134,6 +135,31 @@ func relay(left, right net.Conn, idleTimeout time.Duration) error { return errors.Join(relayErrors...) } +// watchRelayOwner binds full socket closure to the original physical/request +// lifetime, never its completed setup timeout. Register before replaying any +// buffered payload. The returned function stops an unused callback or joins all +// Close work if cancellation already started. Close calls run outside locks. +func watchRelayOwner(owner context.Context, left, right net.Conn) func() { + ownerClosed := make(chan struct{}) + stopOwner := context.AfterFunc(owner, func() { + defer close(ownerClosed) + _ = right.Close() + _ = left.Close() + }) + return func() { + if !stopOwner() { + <-ownerClosed + } + } +} + +// relayWithOwner preserves ordinary half-close and joins both copy pumps as +// well as an already-started cancellation callback before returning. +func relayWithOwner(owner context.Context, left, right net.Conn, idleTimeout time.Duration) error { + defer watchRelayOwner(owner, left, right)() + return relay(left, right, idleTimeout) +} + // relayActivity makes read inactivity a property of the whole tunnel, not // either direction independently. Downloads, uploads and server-push streams // may legitimately have no reverse-direction application data for minutes. diff --git a/internal/proxy/relay_owner_test.go b/internal/proxy/relay_owner_test.go new file mode 100644 index 0000000..8cc0ff8 --- /dev/null +++ b/internal/proxy/relay_owner_test.go @@ -0,0 +1,782 @@ +package proxy + +import ( + "bufio" + "bytes" + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +type proxyHalfCloseAuditConn struct { + net.Conn + closed chan struct{} + once sync.Once + setupContext context.Context +} + +func (c *proxyHalfCloseAuditConn) Close() error { + err := c.Conn.Close() + c.once.Do(func() { close(c.closed) }) + return err +} + +func (c *proxyHalfCloseAuditConn) CloseWrite() error { + return c.Conn.(interface{ CloseWrite() error }).CloseWrite() +} + +type proxyHalfCloseAuditPhase struct { + release chan struct{} + releaseOnce sync.Once + upload chan []byte + originErr chan error + handlerDone chan struct{} + handlerStarted atomic.Bool + upstream chan *proxyHalfCloseAuditConn +} + +func (p *proxyHalfCloseAuditPhase) unblock() { p.releaseOnce.Do(func() { close(p.release) }) } + +type proxyHalfCloseAuditFixture struct { + proxy *net.TCPListener + origin *net.TCPListener + tracker *connTracker + shutdown func(context.Context) error + serveDone chan struct{} + serveErr error + originDone chan struct{} + phases [2]*proxyHalfCloseAuditPhase + mu sync.Mutex + closed bool + owned []net.Conn + prefix *relayOwnerPrefixWriteConn +} + +func (f *proxyHalfCloseAuditFixture) own(c net.Conn) bool { + f.mu.Lock() + if f.closed { + f.mu.Unlock() + _ = c.Close() + return false + } + f.owned = append(f.owned, c) + f.mu.Unlock() + return true +} + +func proxyHalfCloseAuditWait(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 join", label) + return false + } +} + +func newProxyHalfCloseAuditFixture(t *testing.T, mode string, response []byte, bufferedPrefix bool) *proxyHalfCloseAuditFixture { + t.Helper() + f := &proxyHalfCloseAuditFixture{serveDone: make(chan struct{}), originDone: make(chan struct{})} + for i := range f.phases { + f.phases[i] = &proxyHalfCloseAuditPhase{release: make(chan struct{}), upload: make(chan []byte, 1), originErr: make(chan error, 1), handlerDone: make(chan struct{}), upstream: make(chan *proxyHalfCloseAuditConn, 1)} + } + var err error + f.origin, err = net.ListenTCP("tcp4", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + f.proxy, err = net.ListenTCP("tcp4", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + _ = f.origin.Close() + t.Fatal(err) + } + if bufferedPrefix { + f.prefix = &relayOwnerPrefixWriteConn{entered: make(chan []byte, 1), closed: make(chan struct{}), release: make(chan struct{})} + } + started := false + t.Cleanup(func() { + // Independent cleanup releases gates and every actual owned socket, + // even when the production shutdown oracle fails before either join. + for _, p := range f.phases { + p.unblock() + } + if f.prefix != nil { + f.prefix.unblock() + } + _ = f.origin.Close() + _ = f.proxy.Close() + f.mu.Lock() + f.closed = true + owned := append([]net.Conn(nil), f.owned...) + f.mu.Unlock() + for _, c := range owned { + _ = c.Close() + } + if f.shutdown != nil { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _ = f.shutdown(ctx) + } + if started { + proxyHalfCloseAuditWait(t, f.originDone, "origin accept/handler worker cleanup") + serveJoined := proxyHalfCloseAuditWait(t, f.serveDone, "frontend accept/Serve cleanup") + for i, p := range f.phases { + if p.handlerStarted.Load() { + proxyHalfCloseAuditWait(t, p.handlerDone, fmt.Sprintf("handler %d cleanup", i)) + } + } + if serveJoined && f.serveErr != nil { + t.Errorf("frontend Serve = %v", f.serveErr) + } + } + }) + var dialIndex atomic.Int32 + dialer := transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + if network != "tcp" || address != f.origin.Addr().String() { + return nil, fmt.Errorf("non-owned target %s/%s", network, address) + } + index := int(dialIndex.Add(1) - 1) + if index >= len(f.phases) { + return nil, errors.New("unexpected extra dial") + } + c, err := (&net.Dialer{}).DialContext(ctx, network, address) + if err != nil { + return nil, err + } + if bufferedPrefix && index == 1 { + f.prefix.Conn = c + c = f.prefix + } + observed := &proxyHalfCloseAuditConn{Conn: c, closed: make(chan struct{}), setupContext: ctx} + if !f.own(observed) { + return nil, net.ErrClosed + } + f.phases[index].upstream <- observed + return observed, nil + }) + cfg := Config{Dialer: dialer, HandshakeTimeout: time.Second, DialTimeout: 100 * time.Millisecond, IdleTimeout: 3 * time.Second, MaxConnections: 1} + if mode == "http_connect" { + s, err := NewHTTPServer(cfg) + if err != nil { + t.Fatal(err) + } + f.tracker, f.shutdown = s.lifecycle.tracker, s.Shutdown + var handlerIndex atomic.Int32 + s.server.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + i := int(handlerIndex.Add(1) - 1) + if i >= len(f.phases) { + http.Error(w, "unexpected handler", 500) + return + } + p := f.phases[i] + p.handlerStarted.Store(true) + defer close(p.handlerDone) + s.ServeHTTP(w, r) + }) + started = true + var listener net.Listener = f.proxy + if bufferedPrefix { + request := fmt.Sprintf("CONNECT %s HTTP/1.1\r\nHost: %s\r\n\r\n", f.origin.Addr(), f.origin.Addr()) + listener = &relayOwnerPrefixListener{Listener: listener, firstReadBytes: len(request) + len("owned-halfclose-marker")} + } + go func() { defer close(f.serveDone); f.serveErr = s.Serve(listener) }() + } else { + s, err := NewSOCKS5Server(cfg) + if err != nil { + t.Fatal(err) + } + f.tracker, f.shutdown = s.lifecycle.tracker, s.Shutdown + managed, err := s.lifecycle.manage(f.proxy) + if err != nil { + t.Fatal(err) + } + started = true + go func() { + defer close(f.serveDone) + for i := 0; ; i++ { + c, err := managed.Accept() + if err != nil { + if !errors.Is(err, net.ErrClosed) { + f.serveErr = err + } + return + } + if !f.own(c) { + return + } + if i >= len(f.phases) { + _ = c.Close() + f.serveErr = errors.New("unexpected extra accept") + return + } + p := f.phases[i] + p.handlerStarted.Store(true) + go func() { defer close(p.handlerDone); s.serveConn(c) }() + } + }() + } + go func() { + defer close(f.originDone) + for index, p := range f.phases { + c, err := f.origin.AcceptTCP() + if err != nil { + return + } + if !f.own(c) { + return + } + _ = c.SetDeadline(time.Now().Add(5 * time.Second)) + marker := make([]byte, len("owned-halfclose-marker")) + _, err = io.ReadFull(c, marker) + if err == nil { + _, err = c.Write([]byte("ack")) + } + var tail []byte + if err == nil { + tail, err = io.ReadAll(c) + } + p.upload <- append(marker, tail...) + if err != nil { + p.originErr <- err + _ = c.Close() + continue + } + // This gate holds only the ordinary destination's response, not + // any Close callback or frontend worker completion. + <-p.release + if index == 0 { + _, err = c.Write(response) + if err == nil { + err = c.CloseWrite() + } + } + p.originErr <- err + _ = c.Close() + } + }() + return f +} + +func proxyHalfCloseAuditOpen(t *testing.T, f *proxyHalfCloseAuditFixture, mode string) (*net.TCPConn, *bufio.Reader) { + t.Helper() + c, err := net.DialTCP("tcp4", nil, f.proxy.Addr().(*net.TCPAddr)) + if err != nil { + t.Fatal(err) + } + if !f.own(c) { + t.Fatal("fixture already closed") + } + _ = c.SetDeadline(time.Now().Add(2 * time.Second)) + r := bufio.NewReader(c) + if mode == "http_connect" { + _, err = fmt.Fprintf(c, "CONNECT %s HTTP/1.1\r\nHost: %s\r\n\r\n", f.origin.Addr(), f.origin.Addr()) + if err != nil { + t.Fatal(err) + } + response, err := http.ReadResponse(r, &http.Request{Method: http.MethodConnect}) + if err != nil || response.StatusCode != http.StatusOK { + t.Fatalf("CONNECT response = %v / %v", response, err) + } + } else { + socksGreeting(t, c, nil) + host, port := splitAddress(t, f.origin.Addr().String()) + mustWrite(t, c, domainSOCKSRequest(socksCommandConnect, host, port)) + if reply := readSOCKSReply(t, c); reply != socksReplySucceeded { + t.Fatalf("SOCKS reply = %d", reply) + } + } + mustWrite(t, c, []byte("owned-halfclose-marker")) + ack := make([]byte, 3) + if _, err := io.ReadFull(r, ack); err != nil || string(ack) != "ack" { + t.Fatalf("real ack = %q / %v", ack, err) + } + if err := c.CloseWrite(); err != nil { + t.Fatal(err) + } + return c, r +} + +func proxyHalfCloseAuditUpload(t *testing.T, p *proxyHalfCloseAuditPhase) { + t.Helper() + select { + case payload := <-p.upload: + if string(payload) != "owned-halfclose-marker" { + t.Fatalf("actual origin upload = %q", payload) + } + case <-time.After(2 * time.Second): + t.Fatal("origin did not observe complete upload and real EOF") + } +} + +func TestProxyCONNECTOwnerShutdownReleasesHalfClosedUpstream(t *testing.T) { + for _, mode := range []string{"http_connect", "socks_connect"} { + t.Run(mode, func(t *testing.T) { + response := bytes.Repeat([]byte("delayed-origin-response-"), 8192) + f := newProxyHalfCloseAuditFixture(t, mode, response, false) + c, r := proxyHalfCloseAuditOpen(t, f, mode) + proxyHalfCloseAuditUpload(t, f.phases[0]) + select { + case <-f.phases[0].handlerDone: + t.Fatal("healthy handler ended before delayed response") + default: + } + var firstUpstream *proxyHalfCloseAuditConn + select { + case firstUpstream = <-f.phases[0].upstream: + case <-time.After(time.Second): + t.Fatal("missing healthy setup context") + } + if !errors.Is(firstUpstream.setupContext.Err(), context.Canceled) { + t.Fatalf("successful setup context was not detached/canceled: %v", firstUpstream.setupContext.Err()) + } + setupDeadline, ok := firstUpstream.setupContext.Deadline() + if !ok { + t.Fatal("successful setup had no configured deadline") + } + // Cross the original setup deadline before releasing the delayed + // response. Only the physical owner may govern this live relay. + deadlinePassed := time.NewTimer(time.Until(setupDeadline.Add(20 * time.Millisecond))) + <-deadlinePassed.C + f.phases[0].unblock() + got, err := io.ReadAll(r) + if err != nil || !bytes.Equal(got, response) { + t.Fatalf("delayed half-close response %d/%v, want %d exact bytes", len(got), err, len(response)) + } + if !proxyHalfCloseAuditWait(t, f.phases[0].handlerDone, "healthy half-close handler") { + t.FailNow() + } + _ = c.Close() + t.Logf("healthy control: setup deadline crossed + real upload EOF + delayed %d-byte complete response + handler join", len(response)) + + c, r = proxyHalfCloseAuditOpen(t, f, mode) + proxyHalfCloseAuditUpload(t, f.phases[1]) + var upstream *proxyHalfCloseAuditConn + select { + case upstream = <-f.phases[1].upstream: + case <-time.After(time.Second): + t.Fatal("missing owned upstream witness") + } + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond) + defer cancel() + err = f.shutdown(ctx) + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("forced Shutdown = %v, want deadline", err) + } + f.tracker.mu.Lock() + slots := len(f.tracker.conns) + f.tracker.mu.Unlock() + if slots != 0 { + t.Errorf("forced Shutdown retained %d downstream slots", slots) + } + _ = c.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) + _, err = r.ReadByte() + if !errors.Is(err, io.EOF) { + t.Errorf("downstream after forced Shutdown = %v, want real EOF", err) + } + upstreamClosed, handlerJoined := false, false + closed, joined := upstream.closed, f.phases[1].handlerDone + timer := time.NewTimer(200 * time.Millisecond) + observe: + for !upstreamClosed || !handlerJoined { + select { + case <-closed: + upstreamClosed = true + closed = nil + case <-joined: + handlerJoined = true + joined = nil + case <-timer.C: + break observe + } + } + timer.Stop() + t.Logf("before independent cleanup: Shutdown=%v downstreamEOF=%v tracker=%d upstreamCloseReturned=%v handlerJoined=%v idleBudget=3s", ctx.Err(), errors.Is(err, io.EOF), slots, upstreamClosed, handlerJoined) + if !upstreamClosed { + t.Error("forced Shutdown left owned upstream open after inbound pump had already half-closed") + } + if !handlerJoined { + t.Error("forced Shutdown left relay handler blocked on destination reverse Read") + } + // Cleanup recovery is a separately identified positive control, + // never evidence for the preceding forced shutdown assertions. + f.phases[1].unblock() + if !proxyHalfCloseAuditWait(t, f.phases[1].handlerDone, "handler after independent origin release") { + t.FailNow() + } + if !proxyHalfCloseAuditWait(t, upstream.closed, "upstream Close after independent origin release") { + t.FailNow() + } + if !proxyHalfCloseAuditWait(t, f.serveDone, "Serve after Shutdown") { + t.FailNow() + } + if !proxyHalfCloseAuditWait(t, f.originDone, "origin worker after independent release") { + t.FailNow() + } + for i, p := range f.phases { + select { + case originErr := <-p.originErr: + if originErr != nil { + t.Errorf("origin %d = %v", i, originErr) + } + case <-time.After(time.Second): + t.Errorf("origin %d missing completion", i) + } + } + t.Log("independent origin release recovered upstream close, both handlers, Serve/accept and origin joins") + }) + } +} + +type relayOwnerReadWitnessConn struct { + net.Conn + readStarted chan struct{} + startOnce sync.Once + readEnded chan struct{} + readOnce sync.Once +} + +func (c *relayOwnerReadWitnessConn) Read(p []byte) (int, error) { + c.startOnce.Do(func() { close(c.readStarted) }) + n, err := c.Conn.Read(p) + if err != nil { + c.readOnce.Do(func() { close(c.readEnded) }) + } + return n, err +} + +func (c *relayOwnerReadWitnessConn) CloseWrite() error { + return c.Conn.(*net.TCPConn).CloseWrite() +} + +type relayOwnerCloseCompletionConn struct { + *relayOwnerReadWitnessConn + firstClose atomic.Bool + rawClosed chan struct{} + release chan struct{} +} + +func (c *relayOwnerCloseCompletionConn) Close() error { + first := c.firstClose.CompareAndSwap(false, true) + err := c.Conn.Close() + if first { + // The real descriptor has closed before this synthetic completion + // gate. Later Close calls return immediately, not via a blocking Once. + close(c.rawClosed) + <-c.release + } + return err +} + +// relayOwnerTCPPair joins its sole Accept worker before returning. Independent +// cleanup closes the listener and both returned sockets on every error path. +func relayOwnerTCPPair(t *testing.T) (net.Conn, net.Conn) { + t.Helper() + listener, err := net.ListenTCP("tcp4", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + type acceptedResult struct { + conn net.Conn + err error + } + result := make(chan acceptedResult, 1) + joined := make(chan struct{}) + var server, client net.Conn + t.Cleanup(func() { + _ = listener.Close() + if client != nil { + _ = client.Close() + } + if proxyHalfCloseAuditWait(t, joined, "real TCP pair Accept cleanup") { + select { + case pending := <-result: + if pending.conn != nil { + _ = pending.conn.Close() + } + default: + } + } + if server != nil { + _ = server.Close() + } + }) + go func() { + defer close(joined) + conn, err := listener.AcceptTCP() + result <- acceptedResult{conn: conn, err: err} + }() + client, err = net.DialTimeout("tcp4", listener.Addr().String(), time.Second) + if err != nil { + t.Fatal(err) + } + if !proxyHalfCloseAuditWait(t, joined, "real TCP pair Accept") { + t.FailNow() + } + accepted := <-result // The joined worker published this buffered result. + if accepted.err != nil { + t.Fatal(accepted.err) + } + server = accepted.conn + _ = listener.Close() + return server, client +} + +func TestRelayOwnerCancellationJoinsStartedCloseCallback(t *testing.T) { + leftRaw, leftPeer := relayOwnerTCPPair(t) + rightRaw, rightPeer := relayOwnerTCPPair(t) + left := &relayOwnerReadWitnessConn{Conn: leftRaw, readStarted: make(chan struct{}), readEnded: make(chan struct{})} + right := &relayOwnerCloseCompletionConn{ + relayOwnerReadWitnessConn: &relayOwnerReadWitnessConn{Conn: rightRaw, readStarted: make(chan struct{}), readEnded: make(chan struct{})}, + rawClosed: make(chan struct{}), release: make(chan struct{}), + } + release := sync.OnceFunc(func() { close(right.release) }) + owner, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + joined := make(chan struct{}) + t.Cleanup(func() { + release() + cancel() + // Bypass the completion wrapper during independent fixture cleanup. + for _, conn := range []net.Conn{leftRaw, rightRaw, leftPeer, rightPeer} { + _ = conn.Close() + } + proxyHalfCloseAuditWait(t, joined, "relay cancellation worker cleanup") + }) + go func() { + defer close(joined) + result <- relayWithOwner(owner, left, right, 3*time.Second) + }() + if !proxyHalfCloseAuditWait(t, left.readStarted, "upload actual Read entry") || + !proxyHalfCloseAuditWait(t, right.readStarted, "reverse actual Read entry") { + t.FailNow() + } + cancel() + if !proxyHalfCloseAuditWait(t, right.rawClosed, "owner callback actual TCP descriptor close") { + t.FailNow() + } + if err := rightPeer.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + if _, err := rightPeer.Read(make([]byte, 1)); !errors.Is(err, io.EOF) { + t.Fatalf("real target peer after callback descriptor close = %v, want EOF", err) + } + // The reverse pump's real read error triggers the ordinary hard-error + // shutdown, whose repeated Close returns without waiting on the gate. + // Both raw Read calls must have ended, yet the callback Close has not. + if !proxyHalfCloseAuditWait(t, right.readEnded, "reverse raw read cancellation") || + !proxyHalfCloseAuditWait(t, left.readEnded, "upload raw read cancellation") { + t.FailNow() + } + select { + case <-joined: + t.Error("relay returned before its already-started owner Close callback completed") + case <-time.After(75 * time.Millisecond): + } + release() + if !proxyHalfCloseAuditWait(t, joined, "relay after independent Close completion release") { + t.FailNow() + } + if err := <-result; err != nil { + t.Errorf("relay after cancellation = %v", err) + } +} + +type relayOwnerPrefixListener struct { + net.Listener + firstReadBytes int +} + +func (l *relayOwnerPrefixListener) Accept() (net.Conn, error) { + c, err := l.Listener.Accept() + if err != nil { + return nil, err + } + return &relayOwnerPrefixReadConn{Conn: c, firstReadBytes: l.firstReadBytes}, nil +} + +type relayOwnerPrefixReadConn struct { + net.Conn + firstReadBytes int + firstReadDone bool +} + +func (c *relayOwnerPrefixReadConn) Read(p []byte) (int, error) { + if !c.firstReadDone { + c.firstReadDone = true + if len(p) < c.firstReadBytes { + return 0, errors.New("fixture first Read buffer too small") + } + // Coalesce the real CONNECT header and early payload into one actual + // socket read so net/http necessarily buffers the payload. This is a + // fixture scheduling control, not a production reader or TCP claim. + return io.ReadFull(c.Conn, p[:c.firstReadBytes]) + } + return c.Conn.Read(p) +} + +func (c *relayOwnerPrefixReadConn) CloseWrite() error { + return c.Conn.(*net.TCPConn).CloseWrite() +} + +type relayOwnerPrefixWriteConn struct { + net.Conn + entered chan []byte + closed chan struct{} + release chan struct{} + firstWrite sync.Once + closeOnce sync.Once + releaseOnce sync.Once +} + +func (c *relayOwnerPrefixWriteConn) Write(p []byte) (int, error) { + c.firstWrite.Do(func() { + c.entered <- append([]byte(nil), p...) + // Hold only entry to the real prefix Write. This explicit scheduling + // gate is not evidence of naturally occurring kernel backpressure. + select { + case <-c.closed: + case <-c.release: + } + }) + return c.Conn.Write(p) +} + +func (c *relayOwnerPrefixWriteConn) Close() error { + err := c.Conn.Close() + c.closeOnce.Do(func() { close(c.closed) }) + return err +} + +func (c *relayOwnerPrefixWriteConn) CloseWrite() error { + return c.Conn.(*net.TCPConn).CloseWrite() +} + +func (c *relayOwnerPrefixWriteConn) unblock() { + c.releaseOnce.Do(func() { close(c.release) }) +} + +func relayOwnerOpenBufferedPrefix(t *testing.T, f *proxyHalfCloseAuditFixture, healthy bool) (*net.TCPConn, *bufio.Reader) { + t.Helper() + c, err := net.DialTCP("tcp4", nil, f.proxy.Addr().(*net.TCPAddr)) + if err != nil { + t.Fatal(err) + } + if !f.own(c) { + t.Fatal("fixture already closed") + } + _ = c.SetDeadline(time.Now().Add(2 * time.Second)) + request := fmt.Sprintf("CONNECT %s HTTP/1.1\r\nHost: %s\r\n\r\nowned-halfclose-marker", f.origin.Addr(), f.origin.Addr()) + mustWrite(t, c, []byte(request)) + r := bufio.NewReader(c) + response, err := http.ReadResponse(r, &http.Request{Method: http.MethodConnect}) + if err != nil || response.StatusCode != http.StatusOK { + t.Fatalf("early-payload CONNECT response = %v / %v", response, err) + } + if healthy { + ack := make([]byte, 3) + if _, err := io.ReadFull(r, ack); err != nil || string(ack) != "ack" { + t.Fatalf("actual early-prefix ack = %q / %v", ack, err) + } + if err := c.CloseWrite(); err != nil { + t.Fatal(err) + } + } + return c, r +} + +func TestHTTPConnectOwnerShutdownInterruptsBufferedPrefix(t *testing.T) { + response := bytes.Repeat([]byte("delayed-origin-response-"), 8192) + f := newProxyHalfCloseAuditFixture(t, "http_connect", response, true) + c, r := relayOwnerOpenBufferedPrefix(t, f, true) + proxyHalfCloseAuditUpload(t, f.phases[0]) + f.phases[0].unblock() + got, err := io.ReadAll(r) + if err != nil || !bytes.Equal(got, response) { + t.Fatalf("healthy prefix response %d/%v, want %d exact bytes", len(got), err, len(response)) + } + if !proxyHalfCloseAuditWait(t, f.phases[0].handlerDone, "healthy buffered-prefix handler") { + t.FailNow() + } + _ = c.Close() + t.Logf("healthy real CONNECT buffered prefix + upload EOF + %d-byte exact response", len(got)) + + c, r = relayOwnerOpenBufferedPrefix(t, f, false) + select { + case prefix := <-f.prefix.entered: + if string(prefix) != "owned-halfclose-marker" { + t.Fatalf("buffered prefix Write = %q", prefix) + } + case <-time.After(time.Second): + t.Fatal("buffered prefix did not enter owned upstream Write") + } + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond) + defer cancel() + err = f.shutdown(ctx) + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("forced Shutdown during prefix = %v, want deadline", err) + } + f.tracker.mu.Lock() + slots := len(f.tracker.conns) + f.tracker.mu.Unlock() + if slots != 0 { + t.Errorf("forced prefix shutdown retained %d slots", slots) + } + _ = c.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) + _, readErr := r.ReadByte() + if !errors.Is(readErr, io.EOF) { + t.Errorf("forced prefix downstream Read = %v, want EOF", readErr) + } + closed, joined := false, false + closedCh, joinedCh := f.prefix.closed, f.phases[1].handlerDone + timer := time.NewTimer(200 * time.Millisecond) +observe: + for !closed || !joined { + select { + case <-closedCh: + closed = true + closedCh = nil + case <-joinedCh: + joined = true + joinedCh = nil + case <-timer.C: + break observe + } + } + timer.Stop() + t.Logf("before independent prefix release: tracker=%d targetRawClose=%v handlerJoined=%v downstreamEOF=%v", slots, closed, joined, errors.Is(readErr, io.EOF)) + if !closed || !joined { + t.Errorf("owner did not interrupt/join buffered-prefix Write: rawClose=%v handlerJoined=%v", closed, joined) + } + // The synthetic Write gate and ordinary target response gate are released + // independently only after the production lifetime assertions above. + f.prefix.unblock() + f.phases[1].unblock() + if !proxyHalfCloseAuditWait(t, f.phases[1].handlerDone, "prefix handler independent cleanup") || + !proxyHalfCloseAuditWait(t, f.originDone, "prefix origin independent cleanup") || + !proxyHalfCloseAuditWait(t, f.serveDone, "prefix Serve after Shutdown") { + t.FailNow() + } + for i, p := range f.phases { + select { + case originErr := <-p.originErr: + if i == 0 && originErr != nil { + t.Errorf("healthy prefix origin = %v", originErr) + } + if i == 1 && closed && !errors.Is(originErr, io.EOF) { + t.Errorf("prefix origin after actual descriptor Close = %v, want real EOF", originErr) + } + case <-time.After(time.Second): + t.Errorf("prefix origin %d missing completion", i) + } + } +} diff --git a/internal/proxy/socks5.go b/internal/proxy/socks5.go index 3f2578a..23ffd78 100644 --- a/internal/proxy/socks5.go +++ b/internal/proxy/socks5.go @@ -136,13 +136,14 @@ func (s *SOCKS5Server) serveConn(client net.Conn) { } func (s *SOCKS5Server) serveConnect(client net.Conn, request socksRequest) { - ctx := context.Background() + owner := context.Background() 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) + // Bind this caller's setup wait and established relay to full socket + // close, not to a half-close or the independent setup deadline. Do not + // add a payload-consuming reader or close the shared upstream dialer. + owner = tracked.connectionContext(owner) } + ctx := owner var cancel context.CancelFunc if s.cfg.dialTimeout > 0 { ctx, cancel = context.WithTimeout(ctx, s.cfg.dialTimeout) @@ -180,7 +181,7 @@ func (s *SOCKS5Server) serveConnect(client net.Conn, request socksRequest) { return } _ = client.SetWriteDeadline(time.Time{}) - _ = relay(client, upstream, s.cfg.idleTimeout) + _ = relayWithOwner(owner, client, upstream, s.cfg.idleTimeout) } func (s *SOCKS5Server) serveUDPAssociate(client net.Conn, request socksRequest) { diff --git a/internal/tunnel/web_h2_client.go b/internal/tunnel/web_h2_client.go index c5dfca5..fc19d68 100644 --- a/internal/tunnel/web_h2_client.go +++ b/internal/tunnel/web_h2_client.go @@ -374,6 +374,14 @@ func (c *WebH2Client) DialContext(ctx context.Context, network, address string) fail() return nil, net.ErrClosed } + // Detaching the opening watcher is not the ownership handoff. A caller + // canceled while queued on mu still owns only this request, not its session. + if err := contextError(ctx); err != nil { + c.mu.Unlock() + _ = response.Body.Close() + fail() + return nil, err + } c.selected = true session.active++ c.mu.Unlock() diff --git a/internal/tunnel/web_h2_stream_handoff_test.go b/internal/tunnel/web_h2_stream_handoff_test.go new file mode 100644 index 0000000..fed322a --- /dev/null +++ b/internal/tunnel/web_h2_stream_handoff_test.go @@ -0,0 +1,206 @@ +package tunnel + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "net" + "net/http" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +// Deadline returns the real context's deadline snapshot. The first call after +// the fresh physical session becomes Ready occurs inside contextError, after +// it already read a nil Cause and after the caller watcher was stopped. Hold +// the client mutex at that point so real caller cancellation precedes the +// serialized handoff. Done and Value remain the real context's methods; +// neither cancellation publication nor a transport response is fabricated. +type webH2TCPHandoffSnapshot struct { + session *webH2ClientSession + state tls.ConnectionState + opening, active int + selected bool +} + +type webH2TCPHandoffContext struct { + context.Context + client *WebH2Client + entered chan webH2TCPHandoffSnapshot + release <-chan struct{} + fixture context.Context + gated atomic.Bool + expired atomic.Bool +} + +func (c *webH2TCPHandoffContext) Deadline() (time.Time, bool) { + deadline, ok := c.Context.Deadline() + c.client.mu.Lock() + session := c.client.current + ready := session != nil && session.authState == webH2ClientAuthReady && session.auth != nil + if ready && c.gated.CompareAndSwap(false, true) { + c.entered <- webH2TCPHandoffSnapshot{session: session, state: session.conn.ConnectionState(), + opening: session.opening, active: session.active, selected: c.client.selected} + select { + case <-c.release: + case <-c.fixture.Done(): + c.expired.Store(true) + } + } + c.client.mu.Unlock() + return deadline, ok +} + +func TestWebH2TCPHandoffHonorsCallerCancellation(t *testing.T) { + for _, profile := range []FingerprintProfile{FingerprintNative, FingerprintChrome133} { + t.Run(string(profile), func(t *testing.T) { + f := newWebH2DialSharingFixture(t) + target := f.target() + serverTLS, clientTLS := testTLSConfigs(t) + var destinations, covers, unexpected atomic.Int32 + server, err := ListenWebH2(WebH2ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, + DialTimeout: time.Second, HandshakeTimeout: time.Second, + Cover: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + covers.Add(1) + http.NotFound(w, r) + }), + Dialer: transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + destinations.Add(1) + if network != "tcp" || address != target { + unexpected.Add(1) + return nil, errors.New("diagnostic refuses a non-owned destination") + } + return (&net.Dialer{Timeout: time.Second}).DialContext(ctx, "tcp4", target) + }), + }) + if err != nil { + t.Fatal(err) + } + type observation struct { + state *webServerConnectionAuth + bearer string + } + observed := make(chan observation, 3) + original := server.server.Handler + server.server.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + joined := make(chan struct{}) + f.register(joined) + defer close(joined) + state, _ := r.Context().Value(webServerConnectionAuthContextKey{}).(*webServerConnectionAuth) + select { + case observed <- observation{state: state, bearer: r.Header.Get("Proxy-Authorization")}: + default: + unexpected.Add(1) + } + original.ServeHTTP(w, r) + }) + serveJoined := make(chan struct{}) + var serveErr error + f.register(serveJoined) + f.checks = append(f.checks, func() { + if serveErr != nil { + t.Errorf("owned H2 Serve: %v", serveErr) + } + }) + f.close = append(f.close, func() { + if err := server.Close(); err != nil { + t.Errorf("owned H2 server Close: %v", err) + } + }) + go func() { defer close(serveJoined); serveErr = server.Serve(f.ctx) }() + client, entropy, raws := f.client(profile, server.Addr().String(), clientTLS) + caller, cancel := context.WithCancelCause(f.ctx) + f.close = append(f.close, func() { cancel(context.Canceled) }) + release := make(chan struct{}) + ungate := sync.OnceFunc(func() { close(release) }) + gate := &webH2TCPHandoffContext{Context: caller, client: client, + entered: make(chan webH2TCPHandoffSnapshot, 1), release: release, fixture: f.ctx} + // Release before any client Close tries to acquire the held mutex. + f.close = append(f.close, ungate) + results, joined := f.dial(client, gate, target) + var snapshot webH2TCPHandoffSnapshot + select { + case snapshot = <-gate.entered: + case <-f.ctx.Done(): + t.Fatal("actual post-proof scheduling checkpoint exceeded fixture budget") + } + physical := f.raw(raws) + session, state, selected := snapshot.session, snapshot.state, snapshot.selected + ready := session != nil && snapshot.opening == 1 && snapshot.active == 0 + cause := errors.New("private caller cancellation in H2 final handoff window") + cancel(cause) + f.wait(caller.Done(), "real caller cancellation published") + ungate() + if !ready || selected || state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != webH2ALPN || len(state.VerifiedChains) == 0 { + t.Errorf("checkpoint Ready/selected/version/ALPN/chains=%t/%t/%x/%s/%d", ready, selected, state.Version, state.NegotiatedProtocol, len(state.VerifiedChains)) + } + f.wait(joined, "canceled public CONNECT caller joined") + got := <-results + if got.conn != nil || got.err != cause { + t.Errorf("post-proof canceled handoff conn=%t err=%v; want nil and exact caller cause", got.conn != nil, got.err) + } + // Run recovery controls even if the old source returns a live stream. + if got.conn != nil { + assertWebH2TCPHandoffEcho(t, got.conn, "late-detached-success") + _ = got.conn.Close() + t.Log("old handoff returned a genuinely live echo stream despite already-published caller cancellation") + } + for round := range 2 { + ctx, stop := context.WithCancelCause(f.ctx) + conn, err := client.DialContext(ctx, "tcp", target) + stop(errors.New("successful H2 caller no longer owns its stream")) + if err != nil { + t.Fatalf("later healthy CONNECT %d: %v", round, err) + } + assertWebH2TCPHandoffEcho(t, conn, fmt.Sprintf("healthy-detached-%d", round)) + _ = conn.Close() + } + client.mu.Lock() + same, readyAfter, owners := client.current == session, session.authState == webH2ClientAuthReady, session.opening+session.active + client.mu.Unlock() + select { + case <-physical.closed: + t.Error("caller handoff cancellation closed the physical H2 wire") + default: + } + var seen [3]observation + for i := range seen { + select { + case seen[i] = <-observed: + case <-f.ctx.Done(): + t.Fatal("actual handler entry observation exceeded fixture budget") + } + } + _, secondShort := parseWebSessionBearer(seen[1].bearer) + _, thirdShort := parseWebSessionBearer(seen[2].bearer) + if !same || !readyAfter || owners != 0 || seen[0].state == nil || seen[0].state != seen[1].state || seen[0].state != seen[2].state || len(seen[0].bearer) < 385 || !secondShort || !thirdShort || entropy.nonceReads.Load() != 1 || destinations.Load() != 3 || covers.Load() != 0 || unexpected.Load() != 0 { + t.Errorf("Ready/same/owners/full-short-short/nonce/dest/cover/unexpected=%t/%t/%d/%d,%t,%t/%d/%d/%d/%d", readyAfter, same, owners, len(seen[0].bearer), secondShort, thirdShort, entropy.nonceReads.Load(), destinations.Load(), covers.Load(), unexpected.Load()) + } + if !gate.gated.Load() || gate.expired.Load() { + t.Error("controlled handoff gate did not complete within its independent budget") + } + t.Log("two complete post-success-cancel echo controls used one verified Ready physical H2 session and full/short/short authentication; owned fixture cleanup joins all workers") + }) + } +} + +func assertWebH2TCPHandoffEcho(t *testing.T, conn net.Conn, payload string) { + t.Helper() + if err := conn.SetDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + if _, err := io.WriteString(conn, payload); err != nil { + t.Fatalf("actual owned echo write: %v", err) + } + got := make([]byte, len(payload)) + if _, err := io.ReadFull(conn, got); err != nil || string(got) != payload { + t.Fatalf("actual owned echo body=%q err=%v, want %q", got, err, payload) + } +} diff --git a/internal/tunnel/web_udp.go b/internal/tunnel/web_udp.go index 82b2c2f..c3119e1 100644 --- a/internal/tunnel/web_udp.go +++ b/internal/tunnel/web_udp.go @@ -538,12 +538,15 @@ func (p *webUDPPacketConn) session(target string) (*webUDPClientSession, error) p.mu.Unlock() session, err := p.openSession(target) - openedSession := session - terminate := false p.mu.Lock() - delete(p.pending, target) if p.closed && session != nil { - terminate = true + // Keep the opening enumerable until its late successful stream and + // reservation have actually been released. Close snapshots pending + // owners and waits for done; publishing early would let it return with + // live resources. terminate also takes p.mu, so cleanup must run unlocked. + p.mu.Unlock() + session.terminate() + p.mu.Lock() session = nil err = net.ErrClosed } else if err == nil { @@ -553,13 +556,11 @@ func (p *webUDPPacketConn) session(target string) (*webUDPClientSession, error) // goroutine may block briefly in removeSession until this unlocks. session.start() } + delete(p.pending, target) pending.session = session pending.err = err close(pending.done) p.mu.Unlock() - if terminate { - openedSession.terminate() - } if err != nil { return nil, err } diff --git a/internal/tunnel/web_udp_late_close_join_test.go b/internal/tunnel/web_udp_late_close_join_test.go new file mode 100644 index 0000000..af23575 --- /dev/null +++ b/internal/tunnel/web_udp_late_close_join_test.go @@ -0,0 +1,335 @@ +package tunnel + +import ( + "bytes" + "context" + "crypto/tls" + "errors" + "fmt" + "net" + "net/http" + "net/netip" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +// The gate is test-only scheduling after an actual authenticated response. It +// does not fabricate TLS, HTTP status/proof, resolver results, or stream I/O. +type webUDPLateCloseJoinFixture struct { + f *webH2DialSharingFixture + c *WebH3Client + h *webTunnelHandler + target netip.AddrPort + entered chan struct{} + requests chan (<-chan struct{}) + verified atomic.Int32 + resolves atomic.Int32 + unexpected atomic.Int32 +} + +func newWebUDPLateCloseJoinFixture(t *testing.T, blockResolve bool) *webUDPLateCloseJoinFixture { + t.Helper() + f := newWebH2DialSharingFixture(t) + x := &webUDPLateCloseJoinFixture{f: f, entered: make(chan struct{}), requests: make(chan (<-chan struct{}), 4)} + echo, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + x.target = echo.LocalAddr().(*net.UDPAddr).AddrPort() + f.close = append(f.close, func() { _ = echo.Close() }) + echoDone := make(chan struct{}) + f.register(echoDone) + var echoErr error + f.checks = append(f.checks, func() { + if echoErr != nil { + t.Errorf("UDP echo: %v", echoErr) + } + }) + go func() { + defer close(echoDone) + buffer := make([]byte, 2048) + for { + _ = echo.SetDeadline(time.Now().Add(7 * time.Second)) + n, from, err := echo.ReadFromUDPAddrPort(buffer) + if err != nil { + if !errors.Is(err, net.ErrClosed) { + echoErr = err + } + return + } + if _, err := echo.WriteToUDPAddrPort(buffer[:n], from); err != nil { + if !errors.Is(err, net.ErrClosed) { + echoErr = err + } + return + } + } + }() + serverTLS, clientTLS := testTLSConfigs(t) + server, err := ListenWebH3(WebH3ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, + HandshakeTimeout: 2 * time.Second, DialTimeout: 2 * time.Second, + MaxUDPSessions: 2, MaxClientUDPSessions: 2, + Cover: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { x.unexpected.Add(1); http.NotFound(w, r) }), + Dialer: transport.DialFunc(func(context.Context, string, string) (net.Conn, error) { + x.unexpected.Add(1) + return nil, errors.New("audit forbids TCP/outside target dial") + }), + UDPResolver: UDPResolverFunc(func(ctx context.Context, address string) ([]netip.AddrPort, error) { + x.resolves.Add(1) + joined := make(chan struct{}) + f.register(joined) + defer close(joined) + if address != x.target.String() { + x.unexpected.Add(1) + return nil, errors.New("audit requires owned numeric target") + } + if blockResolve { + close(x.entered) + select { + case <-ctx.Done(): + return nil, context.Cause(ctx) + case <-f.ctx.Done(): + return nil, context.Cause(f.ctx) + } + } + return []netip.AddrPort{x.target}, nil + }), + }) + if err != nil { + t.Fatal(err) + } + x.h = server.server.Handler.(*webTunnelHandler) + base := server.server.Handler + server.server.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + joined := make(chan struct{}) + f.register(joined) + defer close(joined) + if r.ProtoMajor != 3 || r.Proto != webConnectUDPProtocol || r.TLS == nil || r.TLS.Version != tls.VersionTLS13 { + x.unexpected.Add(1) + } + select { + case x.requests <- joined: + default: + x.unexpected.Add(1) + } + base.ServeHTTP(w, r) + }) + serveDone := make(chan struct{}) + f.register(serveDone) + var serveErr error + f.checks = append(f.checks, func() { + if serveErr != nil { + t.Errorf("H3 Serve: %v", serveErr) + } + }) + f.close = append(f.close, func() { + if err := server.Close(); err != nil && !errors.Is(err, net.ErrClosed) { + t.Errorf("H3 Close: %v", err) + } + }) + go func() { defer close(serveDone); serveErr = server.Serve(f.ctx) }() + clientTLS = clientTLS.Clone() + clientTLS.VerifyConnection = func(state tls.ConnectionState) error { + if len(state.VerifiedChains) == 0 || state.NegotiatedProtocol != "h3" || state.Version != tls.VersionTLS13 { + return errors.New("audit TLS verification incomplete") + } + x.verified.Add(1) + return nil + } + x.c, err = NewWebH3Client(WebH3ClientConfig{ + ServerAddress: server.Addr().String(), Token: webTestToken, TLSConfig: clientTLS, + FingerprintProfile: H3FingerprintNative, DialTimeout: time.Second, HandshakeTimeout: 2 * time.Second, + }) + if err != nil { + t.Fatal(err) + } + f.close = append(f.close, func() { + if err := x.c.Close(); err != nil { + t.Errorf("client Close: %v", err) + } + }) + f.checks = append(f.checks, func() { + if x.unexpected.Load() != 0 || x.verified.Load() != 1 || x.resolves.Load() != 1 { + t.Errorf("unexpected/verified/resolves=%d/%d/%d", x.unexpected.Load(), x.verified.Load(), x.resolves.Load()) + } + if len(x.c.udpSlots) != 0 || len(x.h.udp.slots) != 0 || len(x.h.core.sem) != 0 { + t.Errorf("cleanup slots client/server/streams=%d/%d/%d", len(x.c.udpSlots), len(x.h.udp.slots), len(x.h.core.sem)) + } + t.Log("independent server/socket/handler/resolver/send/Close workers joined; slots=0") + }) + return x +} + +func (x *webUDPLateCloseJoinFixture) packet() *webUDPPacketConn { + x.f.t.Helper() + p, err := x.c.DialPacket(x.f.ctx) + if err != nil { + x.f.t.Fatal(err) + } + packet := p.(*webUDPPacketConn) + x.f.close = append(x.f.close, func() { _ = packet.Close() }) + return packet +} + +func (x *webUDPLateCloseJoinFixture) requestJoin() { + x.f.t.Helper() + select { + case done := <-x.requests: + x.f.wait(done, "actual CONNECT-UDP handler/workers") + case <-x.f.ctx.Done(): + x.f.t.Fatal("missing actual request") + } +} + +func (x *webUDPLateCloseJoinFixture) zeroUsers() { + x.c.mu.Lock() + s := x.c.conns[x.c.conn] + users, ready, live := -1, false, false + if s != nil { + users, ready, live = s.users, s.authState == webH3ClientAuthReady, s.conn.Context().Err() == nil + } + x.c.mu.Unlock() + if users != 0 || !ready || !live || len(x.c.udpSlots) != 0 { + x.f.t.Errorf("postcleanup users/ready/live/udpSlots=%d/%t/%t/%d", users, ready, live, len(x.c.udpSlots)) + } +} + +func TestWebUDPLateSuccessfulOpenCloseJoinsResources(t *testing.T) { + t.Run("late_valid_success", func(t *testing.T) { + x := newWebUDPLateCloseJoinFixture(t, false) + p := x.packet() + observerEntered, observerRelease := make(chan struct{}), make(chan struct{}) + var releaseOnce sync.Once + releaseObserver := func() { releaseOnce.Do(func() { close(observerRelease) }) } + // Registered last: all failure paths release the gate before packet/client cleanup. + t.Cleanup(releaseObserver) + p.setAuthenticationObserver(func(connected bool) { + if !connected { + x.unexpected.Add(1) + } + close(observerEntered) + select { + case <-observerRelease: + case <-x.f.ctx.Done(): + } + }) + sendDone, closeDone := make(chan struct{}), make(chan struct{}) + sendErr, closeErr := make(chan error, 1), make(chan error, 1) + x.f.register(sendDone) + x.f.register(closeDone) + go func() { defer close(sendDone); sendErr <- p.Send([]byte("late-owned-UDP"), x.target.String()) }() + x.f.wait(observerEntered, "actual valid CONNECT-UDP proof accepted") + p.mu.Lock() + pending := p.pending[x.target.String()] + p.mu.Unlock() + if pending == nil { + t.Fatal("real opening was not pending") + } + go func() { defer close(closeDone); closeErr <- p.Close() }() + x.f.wait(p.done, "Packet.Close ownership/cancellation transition") + // Existing releaseSession must acquire c.mu. Holding it here is a synthetic + // cleanup-completion scheduling barrier, not a custom operation ignoring ctx. + x.c.mu.Lock() + locked := true + defer func() { + if locked { + x.c.mu.Unlock() + } + }() + session := x.c.conns[x.c.conn] + if session == nil || session.users != 1 || len(x.c.udpSlots) != 1 || session.authState != webH3ClientAuthReady { + t.Fatalf("before late handoff users/slot/Ready invalid") + } + releaseObserver() + // Actual server workers have observed cancellation/stream closure. The + // reservation release is still held by the client mutex, so neither + // pending completion nor joined Close may yet be published. + x.requestJoin() + select { + case <-pending.done: + t.Error("late opening published completion before reservation/UDP-slot release") + default: + } + select { + case <-closeDone: + t.Errorf("Packet.Close returned before late successful cleanup: physical users=%d UDP slots=%d (cleanup gate still held)", session.users, len(x.c.udpSlots)) + case <-time.After(100 * time.Millisecond): + t.Log("Packet.Close remained blocked while late cleanup ownership was held") + } + select { + case <-sendDone: + t.Error("late Send returned before its cleanup ownership gate") + default: + } + if session.users != 1 || len(x.c.udpSlots) != 1 { + t.Error("cleanup gate did not retain real stream reservation and UDP slot") + } + x.c.mu.Unlock() + locked = false + x.f.wait(sendDone, "late Send cleanup worker") + x.f.wait(closeDone, "Packet.Close worker") + if err := <-sendErr; !errors.Is(err, net.ErrClosed) { + t.Errorf("late Send=%v, want net.ErrClosed", err) + } + if err := <-closeErr; err != nil { + t.Errorf("Packet.Close=%v", err) + } + x.f.wait(pending.done, "late pending completion after real cleanup") + x.zeroUsers() + t.Log("real TLS13/H3 valid proof preceded Close; late successful stream/UDP slot released after gate; Send=net.ErrClosed") + }) + t.Run("normal_success", func(t *testing.T) { + x := newWebUDPLateCloseJoinFixture(t, false) + p := x.packet() + payload := []byte("normal-owned-UDP") + joined, result := make(chan struct{}), make(chan error, 1) + x.f.register(joined) + go func() { + defer close(joined) + if err := p.Send(payload, x.target.String()); err != nil { + result <- err + return + } + got, from, err := p.Receive() + if err == nil && (!bytes.Equal(got, payload) || from != x.target.String()) { + err = fmt.Errorf("UDP echo=%q/%q", got, from) + } + result <- err + }() + x.f.wait(joined, "normal real UDP echo") + if err := <-result; err != nil { + t.Fatal(err) + } + if err := p.Close(); err != nil { + t.Fatal(err) + } + x.requestJoin() + x.zeroUsers() + t.Log("normal successful UDP echo and joined Close control passed") + }) + t.Run("canceled_open_error", func(t *testing.T) { + x := newWebUDPLateCloseJoinFixture(t, true) + p := x.packet() + joined, result := make(chan struct{}), make(chan error, 1) + x.f.register(joined) + go func() { defer close(joined); result <- p.Send([]byte("cancel-owned-UDP"), x.target.String()) }() + x.f.wait(x.entered, "actual resolver before response") + if err := p.Close(); err != nil { + t.Fatal(err) + } + x.f.wait(joined, "canceled Send error worker") + if err := <-result; !errors.Is(err, context.Canceled) { + t.Errorf("canceled Send=%v, want context.Canceled", err) + } + x.requestJoin() + if len(x.c.udpSlots) != 0 { + t.Errorf("canceled-error Close retained UDP slot=%d", len(x.c.udpSlots)) + } + t.Log("canceled resolver/failed opening Close joined without retained UDP slot") + }) +}