endpointsharding: Skip nil child pickers while updating ClientConn state

balancer/endpointsharding/endpointsharding.go

@@ -190,13 +190,18 @@ func (es *endpointSharding) updateState() {
 	defer es.mu.Unlock()
 
 	children := es.children.Load()
+	pickerCount := 0
 	childStates := make([]ChildState, 0, children.Len())
 
 	for _, child := range children.Values() {
 		bw := child.(*balancerWrapper)
 		childState := bw.childState
 		childStates = append(childStates, childState)
 		childPicker := childState.State.Picker
+		if childPicker == nil {
+			continue
+		}
+		pickerCount++
 		switch childState.State.ConnectivityState {
 		case connectivity.Ready:
 			readyPickers = append(readyPickers, childPicker)
@@ -210,6 +215,15 @@ func (es *endpointSharding) updateState() {
 		}
 	}
 
+	// If there are no pickers available yet, return a queuing picker.
+	if pickerCount == 0 {
+		es.cc.UpdateState(balancer.State{
+			ConnectivityState: connectivity.Connecting,
+			Picker:            base.NewErrPicker(balancer.ErrNoSubConnAvailable),
+		})
+		return
+	}
+
 	// Construct the round robin picker based off the aggregated state. Whatever
 	// the aggregated state, use the pickers present that are currently in that
 	// state only.

balancer/endpointsharding/endpointsharding_test.go

@@ -28,19 +28,25 @@ import (
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/balancer"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/grpclog"
 	"google.golang.org/grpc/internal"
+	"google.golang.org/grpc/internal/balancer/stub"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/internal/testutils/roundrobin"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
 	"google.golang.org/grpc/serviceconfig"
+	"google.golang.org/grpc/status"
 
 	testgrpc "google.golang.org/grpc/interop/grpc_testing"
 )
 
+const defaultShortTestTimeout = 100 * time.Millisecond
+
 type s struct {
 	grpctest.Tester
 }
@@ -157,3 +163,66 @@ func (s) TestEndpointShardingBasic(t *testing.T) {
 		t.Fatalf("error in expected round robin: %v", err)
 	}
 }
+
+// TestEndpointShardingStuckConnecting verifies that the endpointsharding policy
+// handles child policies that haven't given a picker update correctly and
+// doesn't panic.
+func (s) TestEndpointShardingStuckConnecting(t *testing.T) {
+	childPolicyName := t.Name()
+	stub.Register(childPolicyName, stub.BalancerFuncs{
+		UpdateClientConnState: func(_ *stub.BalancerData, ccs balancer.ClientConnState) error {
+			t.Logf("Not sending any picker update for resolver state: %v", ccs)
+			return nil
+		},
+	})
+	childLbJSON := json.RawMessage(fmt.Sprintf(`[{%q: {}}]`, childPolicyName))
+	childLBConfig, parseErr := ParseConfig(childLbJSON)
+	if parseErr != nil {
+		t.Fatalf("Failed to parse child LB config: %v", parseErr)
+	}
+	backend1 := stubserver.StartTestService(t, nil)
+	defer backend1.Stop()
+
+	mr := manual.NewBuilderWithScheme("e2e-test")
+	defer mr.Close()
+
+	petiolePolicyName := t.Name() + "-petiole"
+	stub.Register(petiolePolicyName, stub.BalancerFuncs{
+		Init: func(bd *stub.BalancerData) {
+			bd.Data = NewBalancer(bd.ClientConn, bd.BuildOptions)
+		},
+		UpdateClientConnState: func(bd *stub.BalancerData, state balancer.ClientConnState) error {
+			return bd.Data.(balancer.Balancer).UpdateClientConnState(balancer.ClientConnState{
+				BalancerConfig: childLBConfig,
+				ResolverState:  state.ResolverState,
+			})
+		},
+	})
+	json := fmt.Sprintf(`{"loadBalancingConfig": [{"%s":{}}]}`, petiolePolicyName)
+	sc := internal.ParseServiceConfig.(func(string) *serviceconfig.ParseResult)(json)
+	mr.InitialState(resolver.State{
+		Endpoints: []resolver.Endpoint{
+			{Addresses: []resolver.Address{{Addr: backend1.Address}}},
+		},
+		ServiceConfig: sc,
+	})
+
+	cc, err := grpc.NewClient(mr.Scheme()+":///", grpc.WithResolvers(mr), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		log.Fatalf("Failed to dial: %v", err)
+	}
+	defer cc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), defaultShortTestTimeout)
+	defer cancel()
+	client := testgrpc.NewTestServiceClient(cc)
+
+	// Even though the child LB policy hasn't given a picker update, it is
+	// assumed to be in CONNECTING.
+	if _, err := client.EmptyCall(ctx, &testgrpc.Empty{}); status.Code(err) != codes.DeadlineExceeded {
+		t.Fatalf("EmptyCall() = %s, want %s", status.Code(err), codes.DeadlineExceeded)
+	}
+
+	if got, want := cc.GetState(), connectivity.Connecting; got != want {
+		t.Errorf("ClientConn.GetState() = %v, want = %v", got, want)
+	}
+}