Debugging HTTP/2 Flow Control Exhaustion: Analyzing gRPC Stream Stalls in Production — code snippets

snippet-1.go

// Typical goroutine block stack trace found during a pprof capture:
// goroutine 4821 [semacquire, 10 minutes]:
// google.golang.org/grpc/internal/transport.(*controlBuffer).get(0xc0001a23c0, 0x1)
//       /go/pkg/mod/google.golang.org/grpc@v1.64.0/internal/transport/controlbuf.go:394 +0x95
// google.golang.org/grpc/internal/transport.(*http2Server).Write(0xc00021c000, 0xc0001bc300, 0xc0003c2000, 0x0)
//       /go/pkg/mod/google.golang.org/grpc@v1.64.0/internal/transport/http2_server.go:882 +0x14c
// google.golang.org/grpc.(*serverStream).SendMsg(0xc0003c2000, {0x18ba620, 0xc00045e4c0})
//       /go/pkg/mod/google.golang.org/grpc@v1.64.0/stream.go:1422 +0x125
// main.(*ReporterServer).StreamReports(0xc0002ba140, 0xc00028e180, 0xc0003c2000)
//       /workspace/main.go:84 +0xbc

snippet-2.sh

# Get detailed socket statistics for port 50051
ss -t -i -a 'sport = :50051'

# Output analysis:
# State      Recv-Q Send-Q  Local Address:Port   Peer Address:Port
# ESTAB      0      0       10.0.1.5:50051       10.0.2.12:49210
#      cubic wscale:7,7 rto:200 rtt:2.11/0.85 ato:40 mss:1460 rcvspace:14600 ssthresh:10 cwnd:10

snippet-3.sh

# Capture HTTP/2 traffic on port 50051, displaying stream IDs and window sizes
tshark -i eth0 -Y "http2" -T fields \
  -e frame.number \
  -e tcp.stream \
  -e http2.streamid \
  -e http2.type \
  -e http2.length \
  -e http2.window_size \
  -f "tcp port 50051" -c 100

snippet-4.go

package main

import (
	"log"
	"net"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type ReporterServer struct {
	UnimplementedReporterServer
}

// StreamReports streams heavy messages to the client.
func (s *ReporterServer) StreamReports(req *Empty, stream Reporter_StreamReportsServer) error {
	payload := make([]byte, 1024*1024) // 1MB payload to fill default 64KB window instantly

	for i := 1; i <= 100; i++ {
		report := &Report{
			Id:   int32(i),
			Data: payload,
		}

		log.Printf("[Server] Attempting to send report %d...", i)
		
		// This will block once client's HTTP/2 receive window is exhausted
		err := stream.Send(report)
		if err != nil {
			log.Printf("[Server] Send error on item %d: %v", i, err)
			return status.Errorf(codes.Unavailable, "stream interrupted: %v", err)
		}

		log.Printf("[Server] Report %d sent successfully.", i)
	}
	return nil
}

func main() {
	lis, err := net.Listen("tcp", ":50051")
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}

	// Keep default window sizes (64KB) to easily observe the stall
	s := grpc.NewServer()
	RegisterReporterServer(s, &ReporterServer{})

	log.Printf("Server listening on :50051")
	if err := s.Serve(lis); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}

snippet-5.go

package main

import (
	"context"
	"io"
	"log"
	"time"

	"google.golang.org/grpc"
)

func main() {
	conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}
	defer conn.Close()

	client := NewReporterClient(conn)
	stream, err := client.StreamReports(context.Background(), &Empty{})
	if err != nil {
		log.Fatalf("error opening stream: %v", err)
	}

	log.Println("[Client] Starting stream consumption...")
	for {
		report, err := stream.Recv()
		if err == io.EOF {
			log.Println("[Client] Stream finished.")
			break
		}
		if err != nil {
			log.Fatalf("[Client] Recv error: %v", err)
		}

		log.Printf("[Client] Received report %d (%d bytes)", report.Id, len(report.Data))

		// Artificial slowdown: processing takes 2 seconds per message.
		// Since message is 1MB and window is 64KB, server will stall immediately on message 2.
		time.Sleep(2 * time.Second)
	}
}

snippet-6.go

package main

import (
	"log"
	"net"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
)

func main() {
	// Tuning parameters: 4MB stream-level window, 16MB connection-level window.
	// This easily covers 10Gbps pipelines with typical cloud-region latency.
	const (
		initialStreamWindow = 4 * 1024 * 1024
		initialConnWindow   = 16 * 1024 * 1024
	)

	// Keepalives prevent silent connection drop when streams are inactive or blocked
	kp := keepalive.ServerParameters{
		MaxConnectionIdle: 15 * time.Minute,
		Time:              30 * time.Second,
		Timeout:           10 * time.Second,
	}

	lis, err := net.Listen("tcp", ":50051")
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}

	s := grpc.NewServer(
		grpc.InitialWindowSize(initialStreamWindow),
		grpc.InitialConnWindowSize(initialConnWindow),
		grpc.KeepaliveParams(kp),
	)

	// Register handlers and serve...
	_ = s.Serve(lis)
}

snippet-7.go

package main

import (
	"context"
	"log"
	"time"

	"google.golang.org/grpc"
)

func DialHelper(addr string) (*grpc.ClientConn, error) {
	const (
		initialStreamWindow = 4 * 1024 * 1024
		initialConnWindow   = 16 * 1024 * 1024
	)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	return grpc.DialContext(ctx, addr,
		grpc.WithInsecure(),
		grpc.WithInitialWindowSize(initialStreamWindow),
		grpc.WithInitialConnWindowSize(initialConnWindow),
		grpc.WithBlock(),
	)
}

snippet-8.yaml

static_resources:
  listeners:
  - name: grpc_ingress_listener
    address:
      socket_address: { address: 0.0.0.0, port_value: 50051 }
    filter_chains:
    - filters:
      - name: envoy.filters.network.http_connection_manager
        typed_config:
          "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
          stat_prefix: ingress_grpc
          route_config:
            name: local_route
            virtual_hosts:
            - name: local_service
              domains: ["*"]
              routes:
              - match: { prefix: "/" }
                route: { cluster: backend_grpc_cluster }
          http_filters:
          - name: envoy.filters.http.router
            typed_config:
              "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router
          # Downstream HTTP/2 Flow Control settings
          http2_protocol_options:
            initial_stream_window_size: 4194304     # 4MB
            initial_connection_window_size: 16777216 # 16MB
            max_concurrent_streams: 250
  clusters:
  - name: backend_grpc_cluster
    connect_timeout: 0.50s
    type: STRICT_DNS
    lb_policy: ROUND_ROBIN
    typed_extension_protocol_options:
      envoy.extensions.upstreams.http.v3.HttpProtocolOptions:
        "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
        explicit_http_config:
          http2_protocol_options:
            # Upstream HTTP/2 Flow Control settings
            initial_stream_window_size: 4194304     # 4MB
            initial_connection_window_size: 16777216 # 16MB