diff --git a/docs/WEB_COVER.md b/docs/WEB_COVER.md index 3b1a0f7..e3f8f6d 100644 --- a/docs/WEB_COVER.md +++ b/docs/WEB_COVER.md @@ -359,6 +359,20 @@ When the relay name resolves to both address families, H3 interleaves IPv6 and IPv4 candidates and starts them with a short stagger; each candidate retains its own UDP socket so a blackholed first address cannot consume the entire H3 budget before a working family is tried. + +Source builds after v1.0.1 share the result of a cold H3 physical connection +attempt with all callers already waiting on it. A failed handshake does not +make those callers start replacement handshakes one after another. A later +invocation may retry, subject to the existing `web-auto` cooldown policy. +The shared attempt belongs to the client and retains its configured physical +dial timeout; canceling one caller stops only that caller's wait, not the +attempt needed by other callers. If all callers abandon it, the attempt can +continue until its existing timeout or client shutdown. Client Close cancels +and joins owned dialing and session-cleanup workers before returning. This +does not change authentication, wire profiles, timeout defaults or sibling +stream ownership, and is not a general connection-speed or browser-equivalence +claim. The published v1.0.1 binary does not contain this change. + After a failed H3 attempt, new TCP streams avoid repeating the UDP timeout. `--fallback-cooldown` is a base duration; each failure independently selects a retry point within +/-20% so clients do not probe in a fixed synchronized diff --git a/internal/tunnel/web_h3_client.go b/internal/tunnel/web_h3_client.go index d71a400..397b68d 100644 --- a/internal/tunnel/web_h3_client.go +++ b/internal/tunnel/web_h3_client.go @@ -54,15 +54,27 @@ type WebH3Client struct { udpMaxTargets int udpReceiveQueue int - ctx context.Context - cancel context.CancelFunc - mu sync.Mutex - closed bool - selected bool - conn *quic.Conn - client *http3.ClientConn - conns map[*quic.Conn]*webH3ClientSession - dial chan struct{} + ctx context.Context + cancel context.CancelFunc + mu sync.Mutex + closed bool + selected bool + conn *quic.Conn + client *http3.ClientConn + conns map[*quic.Conn]*webH3ClientSession + dial *webH3DialAttempt + workers sync.WaitGroup + closeDone chan struct{} + closeErr error +} + +// webH3DialAttempt publishes one immutable result to every caller that joined +// it. A caller's context only controls its wait, not the shared physical dial. +type webH3DialAttempt struct { + done chan struct{} + conn *quic.Conn + client *http3.ClientConn + err error } // webH3ClientSession owns one UDP socket because the Chrome profile uses a @@ -191,6 +203,7 @@ func NewWebH3Client(config WebH3ClientConfig) (*WebH3Client, error) { ctx: ctx, cancel: cancel, conns: make(map[*quic.Conn]*webH3ClientSession), + closeDone: make(chan struct{}), }, nil } @@ -463,68 +476,80 @@ func (c *WebH3Client) releaseSession(session *webH3ClientSession) { } func (c *WebH3Client) connection(ctx context.Context) (*quic.Conn, *http3.ClientConn, error) { - for { - if err := ctx.Err(); err != nil { - return nil, nil, context.Cause(ctx) - } - c.mu.Lock() - if c.closed { - c.mu.Unlock() - return nil, nil, net.ErrClosed - } - if c.conn != nil && c.client != nil && c.conn.Context().Err() == nil { - conn, client := c.conn, c.client - c.mu.Unlock() - return conn, client, nil - } - if waiting := c.dial; waiting != nil { - c.mu.Unlock() - select { - case <-waiting: - continue - case <-ctx.Done(): - return nil, nil, ctx.Err() - case <-c.ctx.Done(): - return nil, nil, net.ErrClosed - } - } - waiting := make(chan struct{}) - c.dial = waiting + if err := contextError(ctx); err != nil { + return nil, nil, err + } + c.mu.Lock() + if c.closed { c.mu.Unlock() - - dialCtx, cancel := context.WithTimeout(ctx, c.dialTimeout) - stop := context.AfterFunc(c.ctx, cancel) - session, err := c.dialSession(dialCtx) - stop() - cancel() - var conn *quic.Conn - var client *http3.ClientConn - if session != nil { - conn = session.conn - client = session.client - } - - c.mu.Lock() - if c.closed && session != nil { - _ = conn.CloseWithError(0, "") - _ = session.closeResources() - conn = nil - client = nil - err = net.ErrClosed - } else if err == nil { - c.conn = conn - c.client = client - c.conns[conn] = session - go c.watchConnection(session) - } - c.dial = nil - close(waiting) + return nil, nil, net.ErrClosed + } + if c.conn != nil && c.client != nil && c.conn.Context().Err() == nil { + conn, client := c.conn, c.client c.mu.Unlock() - if err != nil { - return nil, nil, fmt.Errorf("tunnel: dial web-cover H3: %w", err) - } return conn, client, nil } + attempt := c.dial + if attempt == nil { + attempt = &webH3DialAttempt{done: make(chan struct{})} + c.dial = attempt + // All additions precede Close's closed gate, so its Wait also joins + // attempts that have not started running yet. + c.workers.Add(1) + go c.runConnectionDial(attempt) + } + c.mu.Unlock() + select { + case <-attempt.done: + case <-ctx.Done(): + return nil, nil, context.Cause(ctx) + case <-c.ctx.Done(): + if err := contextError(ctx); err != nil { + return nil, nil, err + } + return nil, nil, net.ErrClosed + } + if err := contextError(ctx); err != nil { + return nil, nil, err + } + if c.ctx.Err() != nil { + return nil, nil, net.ErrClosed + } + return attempt.conn, attempt.client, attempt.err +} + +func (c *WebH3Client) runConnectionDial(attempt *webH3DialAttempt) { + defer c.workers.Done() + dialCtx, cancel := context.WithTimeout(c.ctx, c.dialTimeout) + session, err := c.dialSession(dialCtx) + cancel() + if err != nil { + err = fmt.Errorf("tunnel: dial web-cover H3: %w", err) + } + var conn *quic.Conn + var client *http3.ClientConn + c.mu.Lock() + if c.closed { + err = net.ErrClosed + } else if err == nil { + conn, client = session.conn, session.client + c.conn, c.client = conn, client + c.conns[conn] = session + c.workers.Add(1) + go c.watchConnection(session) + } + c.mu.Unlock() + // A late successful dial, or any failed result with owned resources, + // must finish cleanup before publishing completion or releasing Close. + if session != nil && conn == nil { + _ = session.conn.CloseWithError(0, "") + _ = session.closeResources() + } + c.mu.Lock() + attempt.conn, attempt.client, attempt.err = conn, client, err + c.dial = nil + close(attempt.done) + c.mu.Unlock() } func (c *WebH3Client) dialSession(ctx context.Context) (*webH3ClientSession, error) { @@ -660,6 +685,7 @@ func interleaveWebH3Addresses(addresses []net.IPAddr) []net.IPAddr { } func (c *WebH3Client) watchConnection(session *webH3ClientSession) { + defer c.workers.Done() <-session.conn.Context().Done() c.mu.Lock() if session.authState == webH3ClientAuthBootstrapping { @@ -741,11 +767,12 @@ func (c *WebH3Client) retire(conn *quic.Conn) { func (c *WebH3Client) Close() error { c.mu.Lock() if c.closed { + done := c.closeDone c.mu.Unlock() - return nil + <-done + return c.closeErr } c.closed = true - c.cancel() sessions := make([]*webH3ClientSession, 0, len(c.conns)) for _, session := range c.conns { session.auth.close() @@ -755,12 +782,16 @@ func (c *WebH3Client) Close() error { c.client = nil c.conns = make(map[*quic.Conn]*webH3ClientSession) c.mu.Unlock() + c.cancel() var connErr error for _, session := range sessions { connErr = errors.Join(connErr, session.conn.CloseWithError(0, "")) connErr = errors.Join(connErr, session.closeResources()) } - return errors.Join(connErr, c.transport.Close()) + c.workers.Wait() + c.closeErr = errors.Join(connErr, c.transport.Close()) + close(c.closeDone) + return c.closeErr } func (c *WebH3Client) SelectedTransport() string { diff --git a/internal/tunnel/web_h3_close_join_test.go b/internal/tunnel/web_h3_close_join_test.go new file mode 100644 index 0000000..edb4641 --- /dev/null +++ b/internal/tunnel/web_h3_close_join_test.go @@ -0,0 +1,373 @@ +package tunnel + +import ( + "bytes" + "context" + "errors" + "io" + "net" + "net/http" + "runtime" + "strings" + "sync" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +// This fixture delays Close's return only after closing the real UDP socket. +// It is an explicit cleanup-completion barrier, not a simulated blocked I/O. +type webH3CloseJoinPacket struct { + net.PacketConn + entered chan struct{} + release <-chan struct{} + returned chan struct{} + once sync.Once + err error +} + +func (p *webH3CloseJoinPacket) Close() error { + p.once.Do(func() { + p.err = p.PacketConn.Close() + close(p.entered) + <-p.release + close(p.returned) + }) + return p.err +} + +func webH3CloseJoinWait(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 webH3CloseJoinMustWait(t *testing.T, done <-chan struct{}, label string) { + t.Helper() + if !webH3CloseJoinWait(t, done, label) { + t.FailNow() + } +} + +func webH3CloseJoinRead[T any](t *testing.T, values <-chan T, label string) T { + t.Helper() + select { + case value := <-values: + return value + case <-time.After(2 * time.Second): + t.Fatalf("%s did not return a result", label) + var zero T + return zero + } +} + +type webH3CloseJoinOrigin struct { + listener *net.TCPListener + mu sync.Mutex + conn *net.TCPConn + closed bool + done chan struct{} + result chan error +} + +func newWebH3CloseJoinOrigin(t *testing.T, payload []byte) *webH3CloseJoinOrigin { + t.Helper() + listener, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + origin := &webH3CloseJoinOrigin{listener: listener, done: make(chan struct{}), result: make(chan error, 1)} + t.Cleanup(func() { + origin.mu.Lock() + origin.closed = true + conn := origin.conn + origin.mu.Unlock() + _ = listener.Close() + if conn != nil { + _ = conn.Close() + } + webH3CloseJoinWait(t, origin.done, "owned TCP origin worker") + }) + go func() { + defer close(origin.done) + if err := listener.SetDeadline(time.Now().Add(2 * time.Second)); err != nil { + origin.result <- err + return + } + conn, err := listener.AcceptTCP() + if err != nil { + origin.result <- err + return + } + origin.mu.Lock() + if origin.closed { + origin.mu.Unlock() + _ = conn.Close() + origin.result <- net.ErrClosed + return + } + origin.conn = conn + origin.mu.Unlock() + defer func() { + _ = conn.Close() + origin.mu.Lock() + origin.conn = nil + origin.mu.Unlock() + }() + if err = conn.SetDeadline(time.Now().Add(2 * time.Second)); err == nil { + var got []byte + got, err = io.ReadAll(conn) + if err == nil && !bytes.Equal(got, payload) { + err = errors.New("owned TCP origin received the wrong complete payload") + } + if err == nil { + _, err = io.Copy(conn, bytes.NewReader(append([]byte("reply:"), got...))) + } + if err == nil { + err = conn.CloseWrite() + } + } + origin.result <- err + }() + return origin +} + +func webH3CloseJoinCall(client *WebH3Client, started chan<- struct{}, result chan<- error, done chan struct{}) { + defer close(done) + started <- struct{}{} + result <- client.Close() +} + +// The first Close waits in workers.Wait; only the already-closed branch waits +// on a channel. Observe this test's second actual caller at that branch before +// starting the noncompletion window, rather than assuming it was scheduled. +func webH3CloseJoinObserveFollower(t *testing.T, first, second <-chan struct{}) { + t.Helper() + timer := time.NewTimer(time.Second) + defer timer.Stop() + ticker := time.NewTicker(time.Millisecond) + defer ticker.Stop() + for { + select { + case <-first: + t.Fatal("first Close returned before packet cleanup completion") + case <-second: + t.Fatal("concurrent Close returned before packet cleanup completion") + default: + } + buffer := make([]byte, 1<<20) + n := runtime.Stack(buffer, true) + if n == len(buffer) { + t.Fatal("owned Close stack observation was truncated") + } + for _, block := range strings.Split(string(buffer[:n]), "\n\n") { + line, _, _ := strings.Cut(block, "\n") + if strings.Contains(line, "[chan receive]") && + strings.Contains(block, "(*WebH3Client).Close(") && + strings.Contains(block, "webH3CloseJoinCall(") { + t.Log("observed own concurrent Close waiting on the shared completion channel") + return + } + } + select { + case <-timer.C: + t.Fatal("second Close did not reach its actual completion wait") + case <-first: + t.Fatal("first Close returned before packet cleanup completion") + case <-second: + t.Fatal("concurrent Close returned before packet cleanup completion") + case <-ticker.C: + } + } +} + +func TestWebH3CloseJoinsRemovedSessionCleanupAndConcurrentCallers(t *testing.T) { + payload := []byte("owned H3 cleanup join with upload FIN") + origin := newWebH3CloseJoinOrigin(t, payload) + target := origin.listener.Addr().String() + serverTLS, clientTLS := testTLSConfigs(t) + server, err := ListenWebH3(WebH3ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, + Cover: http.NotFoundHandler(), + Dialer: transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + if network != "tcp" || address != target { + return nil, errors.New("unexpected owned cleanup-join destination") + } + return (&net.Dialer{}).DialContext(ctx, network, address) + }), + }) + if err != nil { + t.Fatal(err) + } + handler := server.server.Handler + handlerDone := make(chan struct{}) + bearerLength := make(chan int, 1) + server.server.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer close(handlerDone) + bearerLength <- len(r.Header.Get("Proxy-Authorization")) + handler.ServeHTTP(w, r) + }) + serveCtx, cancelServe := context.WithCancel(context.Background()) + serveDone := make(chan struct{}) + serveResult := make(chan error, 1) + go func() { defer close(serveDone); serveResult <- server.Serve(serveCtx) }() + t.Cleanup(func() { + _ = server.Close() + cancelServe() + webH3CloseJoinWait(t, serveDone, "owned H3 Serve worker") + }) + client, err := NewWebH3Client(WebH3ClientConfig{ + ServerAddress: server.Addr().String(), Token: webTestToken, TLSConfig: clientTLS, + }) + if err != nil { + t.Fatal(err) + } + release := make(chan struct{}) + ungate := sync.OnceFunc(func() { close(release) }) + var packet *webH3CloseJoinPacket + var stream net.Conn + var closeWorkers []<-chan struct{} + t.Cleanup(func() { + ungate() + if packet != nil { + _ = packet.PacketConn.Close() + } + if stream != nil { + _ = stream.Close() + } + cleanupDone := make(chan struct{}) + go func() { defer close(cleanupDone); _ = client.Close() }() + webH3CloseJoinWait(t, cleanupDone, "independent client cleanup") + for _, done := range closeWorkers { + webH3CloseJoinWait(t, done, "Close caller cleanup") + } + if packet != nil { + webH3CloseJoinWait(t, packet.returned, "real packet Close completion") + } + workersDone := make(chan struct{}) + go func() { defer close(workersDone); client.workers.Wait() }() + webH3CloseJoinWait(t, workersDone, "physical dial and session watcher cleanup") + }) + dialCtx, cancelDial := context.WithTimeout(context.Background(), 2*time.Second) + defer cancelDial() + stream, err = client.DialContext(dialCtx, "tcp", target) + if err != nil { + t.Fatal(err) + } + if err = stream.SetDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + if _, err = stream.Write(payload); err != nil { + t.Fatal(err) + } + if err = stream.(*webH3Conn).CloseWrite(); err != nil { + t.Fatal(err) + } + reply, err := io.ReadAll(stream) + if err != nil || !bytes.Equal(reply, append([]byte("reply:"), payload...)) { + t.Fatalf("complete healthy H3 reply = %q, %v", reply, err) + } + webH3CloseJoinMustWait(t, origin.done, "successful complete TCP origin") + if err := webH3CloseJoinRead(t, origin.result, "successful TCP origin"); err != nil { + t.Fatal(err) + } + webH3CloseJoinMustWait(t, handlerDone, "successful authenticated CONNECT handler") + if length := webH3CloseJoinRead(t, bearerLength, "full bootstrap credential"); length < 385 || length > 2047 { + t.Fatalf("initial full bootstrap credential length = %d", length) + } + client.mu.Lock() + session := client.conns[client.conn] + if session == nil || session.conn.Context().Err() != nil || session.retired || + session.authState != webH3ClientAuthReady || session.auth == nil || session.users != 1 || + client.client != session.client || client.dial != nil || len(client.conns) != 1 { + client.mu.Unlock() + t.Fatal("healthy exact live authenticated session was not proved before instrumentation") + } + packet = &webH3CloseJoinPacket{ + PacketConn: session.packet, entered: make(chan struct{}), release: release, returned: make(chan struct{}), + } + session.packet = packet + client.mu.Unlock() + client.retire(session.conn) + if err := stream.Close(); err != nil { + t.Fatal(err) + } + webH3CloseJoinMustWait(t, packet.entered, "real packet closed before completion gate") + if packet.err != nil { + t.Fatalf("instrumented real packet Close = %v", packet.err) + } + client.mu.Lock() + removed := client.conns[session.conn] == nil && client.conn == nil && client.client == nil && session.users == 0 + client.mu.Unlock() + if !removed { + t.Fatal("actual watcher did not remove the retired session before blocked resource cleanup") + } + select { + case <-packet.returned: + t.Fatal("packet Close escaped the explicit completion gate") + default: + } + results := make(chan error, 2) + started := make(chan struct{}, 2) + first, second := make(chan struct{}), make(chan struct{}) + closeWorkers = append(closeWorkers, first, second) + go webH3CloseJoinCall(client, started, results, first) + webH3CloseJoinRead(t, started, "first Close entry") + closedDeadline := time.NewTimer(time.Second) + closedPoll := time.NewTicker(time.Millisecond) + defer closedDeadline.Stop() + defer closedPoll.Stop() + for { + client.mu.Lock() + closed := client.closed + client.mu.Unlock() + if closed { + break + } + select { + case <-closedDeadline.C: + t.Fatal("first Close did not reach the real closed gate") + case <-closedPoll.C: + } + } + go webH3CloseJoinCall(client, started, results, second) + webH3CloseJoinRead(t, started, "second Close entry") + webH3CloseJoinObserveFollower(t, first, second) + select { + case <-first: + t.Fatal("first Close returned while removed-session cleanup was still pending") + case <-second: + t.Fatal("concurrent Close returned while removed-session cleanup was still pending") + case <-client.closeDone: + t.Fatal("client published final closure before resource cleanup completed") + case <-time.After(75 * time.Millisecond): + } + ungate() + webH3CloseJoinMustWait(t, first, "first Close") + webH3CloseJoinMustWait(t, second, "concurrent Close") + webH3CloseJoinMustWait(t, packet.returned, "packet cleanup after gate release") + webH3CloseJoinMustWait(t, client.closeDone, "shared client closure completion") + for range 2 { + if err := webH3CloseJoinRead(t, results, "joined Close"); err != nil { + t.Errorf("Close after complete cleanup = %v", err) + } + } + client.mu.Lock() + final := client.dial == nil && len(client.conns) == 0 && client.conn == nil && client.client == nil + client.mu.Unlock() + if !final { + t.Error("joined Close retained a dial or tracked session") + } + _ = server.Close() + cancelServe() + webH3CloseJoinMustWait(t, serveDone, "successful H3 Serve shutdown") + if err := webH3CloseJoinRead(t, serveResult, "joined H3 Serve"); err != nil { + t.Errorf("H3 Serve = %v", err) + } +} diff --git a/internal/tunnel/web_h3_dial_sharing_test.go b/internal/tunnel/web_h3_dial_sharing_test.go new file mode 100644 index 0000000..0b8b122 --- /dev/null +++ b/internal/tunnel/web_h3_dial_sharing_test.go @@ -0,0 +1,929 @@ +package tunnel + +import ( + "bytes" + "context" + "crypto/tls" + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "net/http" + "reflect" + "runtime" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/apernet/quic-go" + "github.com/cppla/autocar/internal/transport" +) + +const webH3DialSharingBudget = time.Second + +type webH3DialSharingJoin struct { + name string + done <-chan struct{} +} + +// All fixture networking is exact-owned IPv4 loopback. Cleanup independently +// cancels callers, closes resources and joins workers; result arrival is not a +// substitute for completion of the goroutine that produced that result. +type webH3DialSharingFixture struct { + t *testing.T + ctx context.Context + cancel context.CancelFunc + serverTLS *tls.Config + clientTLS *tls.Config + closers []func() + checks []func() + mu sync.Mutex + joins []webH3DialSharingJoin +} + +func newWebH3DialSharingFixture(t *testing.T) *webH3DialSharingFixture { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + f := &webH3DialSharingFixture{t: t, ctx: ctx, cancel: cancel} + t.Cleanup(func() { + cancel() + for i := len(f.closers) - 1; i >= 0; i-- { + f.closers[i]() + } + // Resources are closed before taking the final worker snapshot. Caller + // and Serve workers are registered before launch; handler completion is + // also registered at entry and explicitly observed on successful paths. + f.mu.Lock() + joins := append([]webH3DialSharingJoin(nil), f.joins...) + f.mu.Unlock() + deadline := time.NewTimer(2 * time.Second) + defer deadline.Stop() + for _, worker := range joins { + select { + case <-worker.done: + case <-deadline.C: + t.Errorf("cleanup did not join %s", worker.name) + return + } + } + for _, check := range f.checks { + check() + } + }) + f.serverTLS, f.clientTLS = testTLSConfigs(t) + return f +} + +func (f *webH3DialSharingFixture) join(name string, done <-chan struct{}) { + f.mu.Lock() + f.joins = append(f.joins, webH3DialSharingJoin{name: name, done: done}) + f.mu.Unlock() +} + +func (f *webH3DialSharingFixture) wait(done <-chan struct{}, name string) { + f.t.Helper() + select { + case <-done: + case <-f.ctx.Done(): + f.t.Fatalf("%s exceeded independent fixture budget", name) + } +} + +type webH3DialSharingPhase struct{ peer, dcid string } + +type webH3DialSharingUDP struct { + front, back *net.UDPConn + remote *net.UDPAddr + mu sync.Mutex + phases []webH3DialSharingPhase + seen map[string]bool + peer *net.UDPAddr + first chan struct{} + forward bool + pending []byte + unowned int + err error +} + +// With no destination this is a real UDP blackhole. With a destination it +// buffers the first actual datagram until release and then forwards both ways +// to an owned real HTTP/3 server; it never synthesizes protocol responses. +func (f *webH3DialSharingFixture) udp(address, destination string) *webH3DialSharingUDP { + f.t.Helper() + local, err := net.ResolveUDPAddr("udp4", address) + if err != nil { + f.t.Fatal(err) + } + front, err := net.ListenUDP("udp4", local) + if err != nil { + f.t.Fatal(err) + } + u := &webH3DialSharingUDP{front: front, seen: make(map[string]bool), first: make(chan struct{})} + f.closers = append(f.closers, func() { _ = front.Close() }) + if err := front.SetDeadline(time.Now().Add(7 * time.Second)); err != nil { + f.t.Fatal(err) + } + if destination != "" { + u.remote, err = net.ResolveUDPAddr("udp4", destination) + if err != nil || !u.remote.IP.Equal(net.IPv4(127, 0, 0, 1)) { + f.t.Fatalf("non-owned UDP forwarding destination %q: %v", destination, err) + } + u.back, err = net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + f.t.Fatal(err) + } + f.closers = append(f.closers, func() { _ = u.back.Close() }) + if err := u.back.SetDeadline(time.Now().Add(7 * time.Second)); err != nil { + f.t.Fatal(err) + } + joined := make(chan struct{}) + f.join("UDP return forwarder", joined) + go func() { + defer close(joined) + var buffer [65536]byte + for { + n, source, err := u.back.ReadFromUDP(buffer[:]) + if err != nil { + u.noteError(err) + return + } + u.mu.Lock() + peer := u.peer + if source.String() != u.remote.String() { + u.unowned++ + u.mu.Unlock() + continue + } + u.mu.Unlock() + if peer != nil { + if _, err := front.WriteToUDP(buffer[:n], peer); err != nil { + u.noteError(err) + return + } + } + } + }() + } + joined := make(chan struct{}) + f.join("UDP Initial observer", joined) + go func() { + defer close(joined) + var buffer [65536]byte + for { + n, peer, err := front.ReadFromUDP(buffer[:]) + if err != nil { + u.noteError(err) + return + } + u.mu.Lock() + if !peer.IP.Equal(net.IPv4(127, 0, 0, 1)) { + u.unowned++ + u.mu.Unlock() + continue + } + u.peer = peer + // Count public v1 Initial/DCID invariant fields only. Repeated + // packets from one peer/DCID are not new physical attempts. + if n >= 7 && buffer[0]&0xc0 == 0xc0 && buffer[0]&0x30 == 0 && binary.BigEndian.Uint32(buffer[1:5]) == 1 { + length := int(buffer[5]) + if length > 0 && length <= 20 && 6+length < n { + dcid := fmt.Sprintf("%x", buffer[6:6+length]) + key := peer.String() + "/" + dcid + // A live server can change an Initial's destination CID to + // its SCID without a new connection. Forwarding controls + // therefore record the first Initial per peer and separately + // verify server-auth physical identity after success. A true + // blackhole cannot change that DCID; peer/DCID grouping there + // also permits a later socket to reuse an ephemeral port. + if u.back != nil { + key = peer.String() + } + if !u.seen[key] { + u.seen[key] = true + u.phases = append(u.phases, webH3DialSharingPhase{peer.String(), dcid}) + if len(u.phases) == 1 { + close(u.first) + } + } + } + } + forward := u.forward + if u.back != nil && !forward && u.pending == nil { + u.pending = bytes.Clone(buffer[:n]) + } + u.mu.Unlock() + if u.back != nil && forward { + if _, err := u.back.WriteToUDP(buffer[:n], u.remote); err != nil { + u.noteError(err) + return + } + } + } + }() + return u +} + +func (u *webH3DialSharingUDP) noteError(err error) { + if errors.Is(err, net.ErrClosed) { + return + } + u.mu.Lock() + u.err = errors.Join(u.err, err) + u.mu.Unlock() +} + +func (u *webH3DialSharingUDP) release() error { + u.mu.Lock() + u.forward = true + pending := u.pending + u.pending = nil + u.mu.Unlock() + if u.back == nil || len(pending) == 0 { + return errors.New("missing owned first UDP datagram") + } + _, err := u.back.WriteToUDP(pending, u.remote) + return err +} + +func (u *webH3DialSharingUDP) snapshot() ([]webH3DialSharingPhase, int, error) { + u.mu.Lock() + defer u.mu.Unlock() + return append([]webH3DialSharingPhase(nil), u.phases...), u.unowned, u.err +} + +func (f *webH3DialSharingFixture) phaseCount(u *webH3DialSharingUDP, want int) { + f.t.Helper() + phases, unowned, err := u.snapshot() + for i, phase := range phases { + f.t.Logf("actual v1 Initial phase=%d peer=%s dcid=%s", i+1, phase.peer, phase.dcid) + } + if len(phases) != want || unowned != 0 || err != nil { + f.t.Errorf("physical Initial phases=%d want=%d non-owned=%d observer error=%v", len(phases), want, unowned, err) + } +} + +func (f *webH3DialSharingFixture) h3(address string) *WebH3Client { + f.t.Helper() + client, err := NewWebH3Client(WebH3ClientConfig{ + ServerAddress: address, Token: testToken, TLSConfig: f.clientTLS, + FingerprintProfile: H3FingerprintNative, DialTimeout: webH3DialSharingBudget, + HandshakeTimeout: 2 * time.Second, QUICConfig: &quic.Config{Versions: []quic.Version{quic.Version1}}, + }) + if err != nil { + f.t.Fatal(err) + } + f.closers = append(f.closers, func() { _ = client.Close() }) + return client +} + +type webH3DialSharingCaller struct { + result chan webH3DialSharingResult + joined chan struct{} + cancel context.CancelCauseFunc + goroutineID atomic.Uint64 +} + +type webH3DialSharingResult struct { + err error + success bool +} + +func (f *webH3DialSharingFixture) caller(gate <-chan struct{}, entered chan<- struct{}, dial func(context.Context) (net.Conn, error), payload string) *webH3DialSharingCaller { + ctx, cancel := context.WithCancelCause(f.ctx) + c := &webH3DialSharingCaller{result: make(chan webH3DialSharingResult, 1), joined: make(chan struct{}), cancel: cancel} + f.join("dial caller", c.joined) + go func() { + defer close(c.joined) + defer cancel(nil) + var stack [128]byte + n := runtime.Stack(stack[:], false) + fields := strings.Fields(strings.SplitN(string(stack[:n]), "\n", 2)[0]) + if len(fields) >= 2 { + id, err := strconv.ParseUint(fields[1], 10, 64) + if err == nil { + c.goroutineID.Store(id) + } + } + if entered != nil { + entered <- struct{}{} + } + if gate != nil { + select { + case <-gate: + case <-ctx.Done(): + c.result <- webH3DialSharingResult{err: context.Cause(ctx)} + return + } + } + conn, err := dial(ctx) + success := conn != nil && err == nil + if conn != nil { + if success && payload != "" { + err = webH3DialSharingExchange(conn, payload) + } + _ = conn.Close() + } + c.result <- webH3DialSharingResult{err: err, success: success} + }() + return c +} + +func webH3DialSharingExchange(conn net.Conn, payload string) error { + if err := conn.SetDeadline(time.Now().Add(2 * time.Second)); err != nil { + return err + } + if _, err := io.WriteString(conn, payload); err != nil { + return err + } + writer, ok := conn.(interface{ CloseWrite() error }) + if !ok { + return errors.New("real tunnel omitted CloseWrite") + } + if err := writer.CloseWrite(); err != nil { + return err + } + reply, err := io.ReadAll(conn) + if err == nil && string(reply) != "reply:"+payload { + err = fmt.Errorf("real target reply=%q want=%q", reply, "reply:"+payload) + } + return err +} + +func (f *webH3DialSharingFixture) result(c *webH3DialSharingCaller) webH3DialSharingResult { + f.t.Helper() + var r webH3DialSharingResult + select { + case r = <-c.result: + case <-f.ctx.Done(): + f.t.Fatal("dial result exceeded independent fixture budget") + } + f.wait(c.joined, "dial caller after result") + return r +} + +// Identity only: no unexported reflected fields, Interface, unsafe or new +// production types. The same source compiles with the old channel and the new +// result-bearing pointer. Access to c.dial always holds its production mutex. +func webH3DialSharingAttempt(client *WebH3Client) uintptr { + client.mu.Lock() + defer client.mu.Unlock() + v := reflect.ValueOf(client.dial) + if v.IsNil() { + return 0 + } + return v.Pointer() +} + +func webH3DialSharingWaitStacks() map[uint64]string { + buffer := make([]byte, 512<<10) + n := runtime.Stack(buffer, true) + waiters := make(map[uint64]string) + for _, block := range strings.Split(string(buffer[:n]), "\n\n") { + lines := strings.Split(block, "\n") + if len(lines) > 1 && strings.Contains(lines[0], "[select") && strings.Contains(lines[1], "(*WebH3Client).connection(") && strings.Contains(block, "(*webH3DialSharingFixture).caller.func") { + fields := strings.Fields(lines[0]) + if len(fields) >= 2 { + id, err := strconv.ParseUint(fields[1], 10, 64) + if err == nil { + waiters[id] = block + } + } + } + } + return waiters +} + +func (f *webH3DialSharingFixture) joinedAttempt(client *WebH3Client, u *webH3DialSharingUDP, attempt uintptr, followers []*webH3DialSharingCaller) { + f.t.Helper() + timer := time.NewTimer(500 * time.Millisecond) + defer timer.Stop() + tick := time.NewTicker(time.Millisecond) + defer tick.Stop() + for { + waiters := webH3DialSharingWaitStacks() + joined := 0 + for _, follower := range followers { + if id := follower.goroutineID.Load(); id != 0 && waiters[id] != "" { + joined++ + } + } + phases, _, _ := u.snapshot() + if joined == len(followers) && webH3DialSharingAttempt(client) == attempt && len(phases) == 1 { + f.t.Logf("joined first physical attempt: both exact follower goroutine IDs in actual connection-select stacks, same identity, one Initial phase") + return + } + select { + case <-timer.C: + f.t.Fatalf("same in-flight join not proved: exact followers=%d/%d same=%v phases=%d", joined, len(followers), webH3DialSharingAttempt(client) == attempt, len(phases)) + case <-f.ctx.Done(): + f.t.Fatal("same-attempt witness exceeded independent fixture budget") + case <-tick.C: + } + } +} + +func (f *webH3DialSharingFixture) batch(client *WebH3Client, u *webH3DialSharingUDP, dial func(context.Context) (net.Conn, error), payload string) ([]*webH3DialSharingCaller, uintptr) { + f.t.Helper() + callers := []*webH3DialSharingCaller{f.caller(nil, nil, dial, payload+"0")} + f.wait(u.first, "actual leader Initial") + attempt := webH3DialSharingAttempt(client) + if attempt == 0 { + f.t.Fatal("actual Initial without live physical attempt") + } + gate, entered := make(chan struct{}), make(chan struct{}, 2) + var release sync.Once + f.closers = append(f.closers, func() { release.Do(func() { close(gate) }) }) + for i := 1; i < 3; i++ { + callers = append(callers, f.caller(gate, entered, dial, payload+fmt.Sprint(i))) + } + for range 2 { + select { + case <-entered: + case <-f.ctx.Done(): + f.t.Fatal("followers did not enter actual call barrier") + } + } + release.Do(func() { close(gate) }) + f.joinedAttempt(client, u, attempt, callers[1:]) + return callers, attempt +} + +func TestWebH3DialSharingFailedAttemptAndLaterRetry(t *testing.T) { + f := newWebH3DialSharingFixture(t) + u := f.udp("127.0.0.1:0", "") + client := f.h3(u.front.LocalAddr().String()) + dial := func(ctx context.Context) (net.Conn, error) { + return client.DialContext(ctx, "tcp", "owned-dummy.example:443") + } + callers, first := f.batch(client, u, dial, "") + var shared error + for i, caller := range callers { + r := f.result(caller) + if r.success || !errors.Is(r.err, context.DeadlineExceeded) { + t.Errorf("caller %d: success=%v error=%v, want physical deadline", i, r.success, r.err) + } + if i == 0 { + shared = r.err + } else if r.err != shared { + t.Errorf("caller %d lost immutable shared failure identity: got=%v first=%v", i, r.err, shared) + } + } + f.phaseCount(u, 1) + if webH3DialSharingAttempt(client) != 0 { + t.Error("completed failure retained active physical attempt") + } + // A new invocation, after all original callers joined, may independently + // retry. Failure sharing must not become a permanent cached failure. + retry := f.caller(nil, nil, dial, "") + r := f.result(retry) + if r.success || !errors.Is(r.err, context.DeadlineExceeded) || r.err == shared { + t.Errorf("later independent retry success=%v error=%v shared=%v", r.success, r.err, r.err == shared) + } + f.phaseCount(u, 2) + phases, _, _ := u.snapshot() + if len(phases) >= 2 && phases[0].dcid == phases[1].dcid { + t.Error("later independent retry reused the original Initial DCID") + } + t.Logf("first attempt identity=%x; original callers joined before fresh retry", first) +} + +func (f *webH3DialSharingFixture) target(count int) (string, <-chan error) { + f.t.Helper() + listener, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + f.t.Fatal(err) + } + var mu sync.Mutex + var current *net.TCPConn + closed := false + f.closers = append(f.closers, func() { + _ = listener.Close() + mu.Lock() + closed = true + if current != nil { + _ = current.Close() + } + mu.Unlock() + }) + results := make(chan error, count) + joined := make(chan struct{}) + f.join("real TCP destination", joined) + go func() { + defer close(joined) + for range count { + if err := listener.SetDeadline(time.Now().Add(7 * time.Second)); err != nil { + results <- err + return + } + conn, err := listener.AcceptTCP() + if err != nil { + if !errors.Is(err, net.ErrClosed) { + results <- err + } + return + } + mu.Lock() + if closed { + mu.Unlock() + _ = conn.Close() + return + } + current = conn + mu.Unlock() + err = conn.SetDeadline(time.Now().Add(2 * time.Second)) + var payload []byte + if err == nil { + payload, err = io.ReadAll(conn) + } + if err == nil { + _, err = io.Copy(conn, bytes.NewReader(append([]byte("reply:"), payload...))) + } + if err == nil { + err = conn.CloseWrite() + } + _ = conn.Close() + mu.Lock() + current = nil + mu.Unlock() + results <- err + } + }() + return listener.Addr().String(), results +} + +type webH3DialSharingResponse struct { + proof bool + bearerSize int + short bool + proto int + tlsVersion uint16 + auth *webServerConnectionAuth + joined <-chan struct{} +} + +type webH3DialSharingRelay struct { + address string + handler *webTunnelHandler + responses chan webH3DialSharingResponse + dials atomic.Int32 + cover atomic.Int32 +} + +func (f *webH3DialSharingFixture) relay(h3 bool, target string, count int) *webH3DialSharingRelay { + f.t.Helper() + r := &webH3DialSharingRelay{responses: make(chan webH3DialSharingResponse, count)} + dialer := transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + if network != "tcp" || address != target { + return nil, fmt.Errorf("refused non-owned destination %s/%s", network, address) + } + r.dials.Add(1) + return (&net.Dialer{}).DialContext(ctx, network, address) + }) + cover := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + r.cover.Add(1) + http.NotFound(w, nil) + }) + var serve func(context.Context) error + var closeServer func() error + var install func(http.Handler) + if h3 { + server, err := ListenWebH3(WebH3ServerConfig{ + Address: "127.0.0.1:0", Token: testToken, TLSConfig: f.serverTLS, + QUICConfig: &quic.Config{Versions: []quic.Version{quic.Version1}}, + Cover: cover, Dialer: dialer, + }) + if err != nil { + f.t.Fatal(err) + } + r.address, r.handler = server.Addr().String(), server.server.Handler.(*webTunnelHandler) + serve, closeServer = server.Serve, server.Close + install = func(h http.Handler) { server.server.Handler = h } + } else { + server, err := ListenWebH2(WebH2ServerConfig{ + Address: "127.0.0.1:0", Token: testToken, TLSConfig: f.serverTLS, Cover: cover, Dialer: dialer, + }) + if err != nil { + f.t.Fatal(err) + } + r.address, r.handler = server.Addr().String(), server.server.Handler.(*webTunnelHandler) + serve, closeServer = server.Serve, server.Close + install = func(h http.Handler) { server.server.Handler = h } + } + install(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + joined := make(chan struct{}) + f.join("authenticated handler", joined) + defer close(joined) + _, short := parseWebSessionBearer(request.Header.Get("Proxy-Authorization")) + auth, _ := request.Context().Value(webServerConnectionAuthContextKey{}).(*webServerConnectionAuth) + r.handler.ServeHTTP(w, request) + r.responses <- webH3DialSharingResponse{ + proof: w.Header().Get(webAuthResponseHeader) != "", bearerSize: len(request.Header.Get("Proxy-Authorization")), + short: short, proto: request.ProtoMajor, tlsVersion: webRequestTLSVersion(request), auth: auth, joined: joined, + } + })) + f.closers = append(f.closers, func() { _ = closeServer() }) + joined := make(chan struct{}) + f.join("relay Serve", joined) + var serveErr error + go func() { defer close(joined); serveErr = serve(f.ctx) }() + f.checks = append(f.checks, func() { + if serveErr != nil { + f.t.Errorf("owned relay Serve: %v", serveErr) + } + }) + return r +} + +func (f *webH3DialSharingFixture) healthy(relay *webH3DialSharingRelay, target <-chan error, count, proto int) { + f.t.Helper() + var physical *webServerConnectionAuth + full, short := 0, 0 + for range count { + var response webH3DialSharingResponse + select { + case response = <-relay.responses: + case <-f.ctx.Done(): + f.t.Fatal("real authenticated handler did not complete") + } + f.wait(response.joined, "authenticated handler after result") + if !response.proof || response.proto != proto || response.tlsVersion != tls.VersionTLS13 || response.auth == nil { + f.t.Errorf("real authenticated response proof=%v protocol=%d TLS=%x physical=%v", response.proof, response.proto, response.tlsVersion, response.auth != nil) + } + if physical == nil { + physical = response.auth + } else if response.auth != physical { + f.t.Error("concurrent healthy requests replaced physical authenticated connection") + } + if response.short { + short++ + } else { + full++ + if response.bearerSize < 385 { + f.t.Error("bootstrap did not contain the full padded credential") + } + } + select { + case err := <-target: + if err != nil { + f.t.Errorf("owned TCP full payload / FIN: %v", err) + } + case <-f.ctx.Done(): + f.t.Fatal("real TCP target did not complete") + } + } + if full != 1 || short != count-1 || relay.dials.Load() != int32(count) || relay.cover.Load() != 0 || len(relay.handler.core.sem) != 0 { + f.t.Errorf("authenticated healthy control full=%d short=%d destination dials=%d cover=%d slots=%d", full, short, relay.dials.Load(), relay.cover.Load(), len(relay.handler.core.sem)) + } + f.t.Logf("real TLS1.3 protocol=%d healthy CONNECTs=%d: one full bootstrap, %d short continuations, full payload+FIN, joined handlers and zero slots", proto, count, count-1) +} + +func TestWebH3DialCancellationIsolation(t *testing.T) { + for _, canceled := range []int{0, 1} { + name := "initiating_caller" + if canceled == 1 { + name = "joined_follower" + } + t.Run(name, func(t *testing.T) { + f := newWebH3DialSharingFixture(t) + target, targetResults := f.target(2) + relay := f.relay(true, target, 2) + u := f.udp("127.0.0.1:0", relay.address) + client := f.h3(u.front.LocalAddr().String()) + dial := func(ctx context.Context) (net.Conn, error) { return client.DialContext(ctx, "tcp", target) } + callers, attempt := f.batch(client, u, dial, "h3-cancellation-") + cause := errors.New("owned custom caller cancellation") + callers[canceled].cancel(cause) + r := f.result(callers[canceled]) + if r.success || r.err != cause { + t.Errorf("canceled caller lost custom cause: success=%v err=%v want=%v", r.success, r.err, cause) + } + if got := webH3DialSharingAttempt(client); got != attempt { + t.Errorf("caller cancellation replaced shared physical attempt: got=%x first=%x", got, attempt) + } + if err := u.release(); err != nil { + t.Fatal(err) + } + for i, caller := range callers { + if i == canceled { + continue + } + r := f.result(caller) + if !r.success || r.err != nil { + t.Errorf("surviving caller %d real H3 payload: success=%v err=%v", i, r.success, r.err) + } + } + f.phaseCount(u, 1) + f.healthy(relay, targetResults, 2, 3) + }) + } + t.Run("pre_canceled_skips_physical_dial", func(t *testing.T) { + f := newWebH3DialSharingFixture(t) + u := f.udp("127.0.0.1:0", "") + client := f.h3(u.front.LocalAddr().String()) + ctx, cancel := context.WithCancelCause(f.ctx) + cause := errors.New("owned pre-canceled cause") + cancel(cause) + conn, err := client.DialContext(ctx, "tcp", "owned-dummy.example:443") + if conn != nil { + _ = conn.Close() + } + if conn != nil || err != cause || webH3DialSharingAttempt(client) != 0 { + t.Errorf("pre-canceled dial conn=%v error=%v attempt=%x", conn != nil, err, webH3DialSharingAttempt(client)) + } + f.phaseCount(u, 0) + }) +} + +func TestWebH3DialClientCloseJoinsBlackhole(t *testing.T) { + f := newWebH3DialSharingFixture(t) + u := f.udp("127.0.0.1:0", "") + client := f.h3(u.front.LocalAddr().String()) + callers, _ := f.batch(client, u, func(ctx context.Context) (net.Conn, error) { + return client.DialContext(ctx, "tcp", "owned-dummy.example:443") + }, "") + closeJoined := make(chan struct{}) + f.join("client Close", closeJoined) + var closeErr error + go func() { defer close(closeJoined); closeErr = client.Close() }() + f.wait(closeJoined, "Close of active physical blackhole") + if closeErr != nil || webH3DialSharingAttempt(client) != 0 { + t.Errorf("Close returned with error=%v live physical attempt=%x", closeErr, webH3DialSharingAttempt(client)) + } + for i, caller := range callers { + r := f.result(caller) + if r.success || !errors.Is(r.err, net.ErrClosed) { + t.Errorf("caller %d after joined client Close: success=%v error=%v", i, r.success, r.err) + } + } + stacks := make([]byte, 512<<10) + n := runtime.Stack(stacks, true) + if strings.Contains(string(stacks[:n]), "(*WebH3Client).dialSession(") || strings.Contains(string(stacks[:n]), "(*WebH3Client).dialSessionAddresses(") { + t.Error("physical dial worker still running after client Close and caller joins") + } + f.phaseCount(u, 1) +} + +func TestWebH3DialFallbackReachesAuthenticatedH2(t *testing.T) { + f := newWebH3DialSharingFixture(t) + target, targetResults := f.target(4) + relay := f.relay(false, target, 4) + // Same owned address/port: real TCP HTTP/2 relay, UDP blackhole. + u := f.udp(relay.address, "") + client, err := NewWebClient(WebClientConfig{ + ServerAddress: relay.address, Token: testToken, TLSConfig: f.clientTLS, + H3FingerprintProfile: H3FingerprintNative, QUICConfig: &quic.Config{Versions: []quic.Version{quic.Version1}}, + H3DialTimeout: webH3DialSharingBudget, H2DialTimeout: time.Second, + HandshakeTimeout: 2 * time.Second, PrimaryAttemptTimeout: 4 * time.Second, FallbackCooldown: time.Minute, + }) + if err != nil { + t.Fatal(err) + } + f.closers = append(f.closers, func() { _ = client.Close() }) + h3 := client.primary.dialer.(*WebH3Client) + dial := func(ctx context.Context) (net.Conn, error) { return client.DialContext(ctx, "tcp", target) } + callers, _ := f.batch(h3, u, dial, "real-h2-fallback-") + for i, caller := range callers { + r := f.result(caller) + if !r.success || r.err != nil { + t.Errorf("caller %d verified real H2 fallback: success=%v error=%v", i, r.success, r.err) + } + } + f.phaseCount(u, 1) + client.mu.Lock() + cooldown := !client.primaryFailedAt.IsZero() + client.mu.Unlock() + if !cooldown || client.SelectedTransport() != webAuthTransportH2 { + t.Errorf("fallback cooldown=%v selected=%q", cooldown, client.SelectedTransport()) + } + // A later invocation uses the existing cooldown and warm real H2 session, + // not a synthetic fallback. It must not start a new physical UDP attempt. + later := f.caller(nil, nil, dial, "real-h2-cooldown") + if r := f.result(later); !r.success || r.err != nil { + t.Errorf("later real H2 cooldown exchange success=%v error=%v", r.success, r.err) + } + f.phaseCount(u, 1) + f.healthy(relay, targetResults, 4, 2) + t.Log("custom physical budget=1s, primary budget=4s; defaults remain 5s, not a default latency or browser-equivalence claim") +} + +// webH3DialUnpublishedDeadlineContext is a synthetic, fixed-deadline Context +// model. Err and Done delegate to the independently bounded fixture context, +// deliberately holding publication of this deadline until fixture cleanup. +// This models the timer-publication gap; it is not evidence of a naturally +// observed scheduling race. Deadline never changes, and no sleep is an entry +// or completion witness. +type webH3DialUnpublishedDeadlineContext struct { + context.Context + deadline time.Time +} + +func (ctx webH3DialUnpublishedDeadlineContext) Deadline() (time.Time, bool) { + return ctx.deadline, true +} + +func TestWebH3DialElapsedDeadlineBeforePublication(t *testing.T) { + t.Run("expired_entry_skips_physical_dial", func(t *testing.T) { + f := newWebH3DialSharingFixture(t) + u := f.udp("127.0.0.1:0", "") + client := f.h3(u.front.LocalAddr().String()) + ctx := webH3DialUnpublishedDeadlineContext{Context: f.ctx, deadline: time.Now().Add(-time.Second)} + if ctx.Err() != nil { + t.Fatal("synthetic deadline was already published") + } + conn, err := client.DialContext(ctx, "tcp", "owned-dummy.example:443") + if conn != nil { + _ = conn.Close() + } + if conn != nil || err != context.DeadlineExceeded || webH3DialSharingAttempt(client) != 0 { + t.Errorf("expired entry conn=%v error=%v attempt=%x; want exact caller DeadlineExceeded and no dial", conn != nil, err, webH3DialSharingAttempt(client)) + } + f.phaseCount(u, 0) + }) + + t.Run("caller_context_precedes_closed_client", func(t *testing.T) { + f := newWebH3DialSharingFixture(t) + u := f.udp("127.0.0.1:0", "") + client := f.h3(u.front.LocalAddr().String()) + if err := client.Close(); err != nil { + t.Fatal(err) + } + expired := time.Now().Add(-time.Second) + ctx := webH3DialUnpublishedDeadlineContext{Context: f.ctx, deadline: expired} + conn, err := client.DialContext(ctx, "tcp", "owned-dummy.example:443") + if conn != nil { + _ = conn.Close() + } + if conn != nil || err != context.DeadlineExceeded { + t.Errorf("expired closed-client conn=%v error=%v; want exact caller DeadlineExceeded before ErrClosed", conn != nil, err) + } + parent, cancel := context.WithCancelCause(f.ctx) + cause := errors.New("owned elapsed-deadline custom cause") + cancel(cause) + caused := webH3DialUnpublishedDeadlineContext{Context: parent, deadline: expired} + conn, err = client.DialContext(caused, "tcp", "owned-dummy.example:443") + if conn != nil { + _ = conn.Close() + } + if conn != nil || err != cause { + t.Errorf("custom-cause closed-client conn=%v error=%v; want original cause identity", conn != nil, err) + } + f.phaseCount(u, 0) + }) + + t.Run("completed_attempt_preserves_elapsed_caller_priority", func(t *testing.T) { + f := newWebH3DialSharingFixture(t) + u := f.udp("127.0.0.1:0", "") + client := f.h3(u.front.LocalAddr().String()) + ctx := webH3DialUnpublishedDeadlineContext{Context: f.ctx, deadline: time.Now().Add(500 * time.Millisecond)} + result := make(chan webH3DialSharingResult, 1) + joined := make(chan struct{}) + f.join("unpublished-deadline caller", joined) + go func() { + defer close(joined) + conn, err := client.DialContext(ctx, "tcp", "owned-dummy.example:443") + if conn != nil { + _ = conn.Close() + } + result <- webH3DialSharingResult{err: err, success: conn != nil} + }() + f.wait(u.first, "actual Initial before synthetic deadline") + if !time.Now().Before(ctx.deadline) { + t.Fatal("first actual Initial not observed before fixed caller deadline; completion-path witness invalid") + } + client.mu.Lock() + attempt := client.dial + client.mu.Unlock() + if attempt == nil { + t.Fatal("actual Initial had no active physical attempt") + } + f.wait(joined, "caller after physical blackhole completion") + r := <-result + select { + case <-attempt.done: + default: + t.Fatal("caller returned before immutable physical result was published") + } + if time.Now().Before(ctx.deadline) || ctx.Err() != nil { + t.Fatal("fixed deadline not elapsed with Err still pending") + } + select { + case <-ctx.Done(): + t.Fatal("synthetic deadline publication was not held") + default: + } + if r.success || r.err != context.DeadlineExceeded { + t.Errorf("completed caller success=%v error=%v; want exact caller DeadlineExceeded, not wrapped physical timeout", r.success, r.err) + } + if attempt.err == nil || !errors.Is(attempt.err, context.DeadlineExceeded) || attempt.err == r.err { + t.Errorf("immutable physical error=%v caller=%v; want distinct physical timeout and caller deadline identities", attempt.err, r.err) + } + if webH3DialSharingAttempt(client) != 0 { + t.Error("completed physical failure retained active attempt") + } + f.phaseCount(u, 1) + }) +}