Deadlock over unix socket with concurrent unary RPCs on one ClientConn (regression in v1.82.1)

go run . -c 64        # ~660k RPCs, fine
go run . -c 512       # deadlocks within a second, permanently
go run . -tcp -c 512  # ~900k RPCs, fine

client  http2Client.reader        -> controlBuffer.throttle
server  http2Server.HandleStreams -> controlBuffer.throttle
both    loopyWriter               -> blocked in net.(*conn).Write

const maxQueuedTransportResponseFrames = 50
func (*registerStream) isTransportResponseFrame() bool { return false }

latch queued=265 throttled=100 limit=100 : cleanupStream=100 clientHeaders=83 dataFrame=82
latch queued=100 throttled=100 limit=100 : registerStream=98 incomingWindowUpdate=1 ping=1
latch queued=211 throttled=100 limit=100 : cleanupStream=97 clientHeaders=55 dataFrame=56 outgoingWindowUpdate=2 ping=1

// Deadlock repro: one ClientConn over a unix socket + concurrent unary RPCs.
// All grpc options are defaults. A raw []byte codec and a hand-written
// ServiceDesc avoid any codegen, so this is a single self-contained file.
//
//	go run . -c 64    # completes normally
//	go run . -c 512   # deadlocks permanently within a second
package main

import (
	"context"
	"flag"
	"fmt"
	"net"
	"os"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/encoding"
)

var (
	concurrency = flag.Int("c", 512, "concurrent unary RPCs sharing one ClientConn")
	duration    = flag.Duration("d", 20*time.Second, "max run time")
	stallAfter  = flag.Duration("stall", 8*time.Second, "declare deadlock after this long with no completions")
	useTCP      = flag.Bool("tcp", false, "use TCP loopback instead of a unix socket (does not deadlock)")
)

type rawCodec struct{}

func (rawCodec) Marshal(v any) ([]byte, error) { return *(v.(*[]byte)), nil }
func (rawCodec) Unmarshal(d []byte, v any) error {
	b := make([]byte, len(d))
	copy(b, d)
	*(v.(*[]byte)) = b
	return nil
}
func (rawCodec) Name() string { return "raw" }

const svc, method = "repro.Echo", "Call"

func main() {
	flag.Parse()
	encoding.RegisterCodec(rawCodec{})

	payload := make([]byte, 1024)
	sd := grpc.ServiceDesc{
		ServiceName: svc,
		HandlerType: (*any)(nil),
		Methods: []grpc.MethodDesc{{
			MethodName: method,
			Handler: func(_ any, ctx context.Context, dec func(any) error, _ grpc.UnaryServerInterceptor) (any, error) {
				var in []byte
				if err := dec(&in); err != nil {
					return nil, err
				}
				out := payload
				return &out, nil
			},
		}},
	}

	network, addr := "unix", fmt.Sprintf("/tmp/grpc-deadlock-%d.sock", os.Getpid())
	if *useTCP {
		network, addr = "tcp", "127.0.0.1:0"
	} else {
		os.Remove(addr)
		defer os.Remove(addr)
	}
	lis, err := net.Listen(network, addr)
	if err != nil {
		panic(err)
	}
	srv := grpc.NewServer(grpc.ForceServerCodec(rawCodec{}))
	srv.RegisterService(&sd, new(any))
	go srv.Serve(lis)

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultCallOptions(grpc.ForceCodec(rawCodec{})),
	}
	target := lis.Addr().String()
	if !*useTCP {
		sock := addr
		opts = append(opts, grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) {
			var d net.Dialer
			return d.DialContext(ctx, "unix", sock)
		}))
		target = "passthrough:///unix"
	}
	conn, err := grpc.NewClient(target, opts...)
	if err != nil {
		panic(err)
	}

	fmt.Printf("transport=%s concurrency=%d\n", network, *concurrency)

	var ops atomic.Int64
	ctx, cancel := context.WithTimeout(context.Background(), *duration)
	defer cancel()

	var wg sync.WaitGroup
	for i := 0; i < *concurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			req := make([]byte, 1024)
			for ctx.Err() == nil {
				var out []byte
				if err := conn.Invoke(ctx, "/"+svc+"/"+method, &req, &out); err != nil {
					return
				}
				ops.Add(1)
			}
		}()
	}

	go func() {
		var last int64
		lastProgress := time.Now()
		for {
			time.Sleep(500 * time.Millisecond)
			n := ops.Load()
			if n != last {
				last, lastProgress = n, time.Now()
				continue
			}
			if time.Since(lastProgress) > *stallAfter {
				buf := make([]byte, 1<<22)
				buf = buf[:runtime.Stack(buf, true)]
				s := string(buf)
				fmt.Printf("DEADLOCK: no RPC completed for %s (completed=%d)\n",
					time.Since(lastProgress).Truncate(time.Second), n)
				fmt.Printf("  goroutines in controlBuffer.throttle : %d\n", strings.Count(s, "controlBuffer).throttle"))
				fmt.Printf("  goroutines blocked on socket write   : %d\n", strings.Count(s, "internal/poll.(*FD).Write"))
				os.Exit(1)
			}
		}
	}()

	wg.Wait()
	fmt.Printf("completed=%d in %s, no deadlock\n", ops.Load(), *duration)
}