[issue #28771] Fix pick_first policy to clear selected_ when deleting subchannel_list_ (#28824)

* [issue #28771] Fix pick_first policy to clear selected_ when promoting
a pending subchannel list to the active subchannel list when all
subchannels have been attempted and are in an error state.

* address comments

* revert idle_filter

src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc

@@ -301,6 +301,7 @@ void PickFirst::PickFirstSubchannelData::ProcessConnectivityChangeLocked(
   GPR_ASSERT(connectivity_state != GRPC_CHANNEL_SHUTDOWN);
   // Handle updates for the currently selected subchannel.
   if (p->selected_ == this) {
+    GPR_ASSERT(subchannel_list() == p->subchannel_list_.get());
     if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_pick_first_trace)) {
       gpr_log(GPR_INFO,
               "Pick First %p selected subchannel connectivity changed to %s", p,
@@ -409,6 +410,7 @@ void PickFirst::PickFirstSubchannelData::ProcessConnectivityChangeLocked(
                     p, p->latest_pending_subchannel_list_.get(),
                     p->subchannel_list_.get());
           }
+          p->selected_ = nullptr;  // owned by p->subchannel_list_
           p->subchannel_list_ = std::move(p->latest_pending_subchannel_list_);
         }
         // If this is the current subchannel list (either because we were

test/cpp/end2end/client_lb_end2end_test.cc

@@ -798,6 +798,39 @@ TEST_F(ClientLbEnd2endTest, PickFirstUpdateSuperset) {
   EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
 }
 
+TEST_F(ClientLbEnd2endTest, PickFirstUpdateToUnconnected) {
+  const int kNumServers = 2;
+  CreateServers(kNumServers);
+  StartServer(0);
+  auto response_generator = BuildResolverResponseGenerator();
+  auto channel = BuildChannel("pick_first", response_generator);
+  auto stub = BuildStub(channel);
+
+  std::vector<int> ports;
+
+  // Try to send rpcs against a list where the server is available.
+  ports.emplace_back(servers_[0]->port_);
+  response_generator.SetNextResolution(ports);
+  gpr_log(GPR_INFO, "****** SET [0] *******");
+  CheckRpcSendOk(stub, DEBUG_LOCATION);
+
+  // Send resolution for which all servers are currently unavailable. Eventually
+  // this triggers replacing the existing working subchannel_list with the new
+  // currently unresponsive list.
+  ports.clear();
+  ports.emplace_back(grpc_pick_unused_port_or_die());
+  ports.emplace_back(servers_[1]->port_);
+  response_generator.SetNextResolution(ports);
+  gpr_log(GPR_INFO, "****** SET [unavailable] *******");
+  EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
+
+  // Ensure that the last resolution was installed correctly by verifying that
+  // the channel becomes ready once one of if its endpoints becomes available.
+  gpr_log(GPR_INFO, "****** StartServer(1) *******");
+  StartServer(1);
+  EXPECT_TRUE(WaitForChannelReady(channel.get()));
+}
+
 TEST_F(ClientLbEnd2endTest, PickFirstGlobalSubchannelPool) {
   // Start one server.
   const int kNumServers = 1;