[xds_client_fuzzer] fix bug in fake transport (#33115)

Fixes `FakeXdsTransport` to remove itself from the map in
`FakeXdsTransportFactory` when it gets orphaned by the `XdsClient`, so
that a subsequent creation of a new transport for the same server does
not trigger an assertion due to the transport already existing in the
map.

Fixes internal b/259362837.

test/core/xds/xds_client_corpora/assert_entry_is_null

@@ -0,0 +1,26 @@
+bootstrap: "{\"xds_servers\": [{\"server_uri\":\"xds.example.com:443\", \"channel_creds\":[{\"type\": \"fake\"}]}]}"
+actions {
+  start_watch {
+    resource_type {
+      route_config {
+      }
+    }
+  }
+}
+actions {
+  stop_watch {
+    resource_type {
+      route_config {
+      }
+    }
+  }
+}
+actions {
+  start_watch {
+    resource_type {
+      route_config {
+      }
+    }
+    resource_name: "{\"xds_servers\": [\203\2600\027erver_uri\":\"xds\013.example.com:443\", \"channel_creds\":[{\"type\": \"fake\"}]}]}"
+  }
+}

test/core/xds/xds_transport_fake.cc

@@ -169,6 +169,14 @@ void FakeXdsTransportFactory::FakeXdsTransport::TriggerConnectionFailure(
 }
 
 void FakeXdsTransportFactory::FakeXdsTransport::Orphan() {
+  {
+    MutexLock lock(&factory_->mu_);
+    auto it = factory_->transport_map_.find(&server_);
+    if (it != factory_->transport_map_.end() && it->second == this) {
+      factory_->transport_map_.erase(it);
+    }
+  }
+  factory_.reset();
   {
     MutexLock lock(&mu_);
     // Can't destroy on_connectivity_failure_ synchronously, since that
@@ -234,8 +242,8 @@ FakeXdsTransportFactory::Create(
   auto& entry = transport_map_[&server];
   GPR_ASSERT(entry == nullptr);
   auto transport = MakeOrphanable<FakeXdsTransport>(
-      std::move(on_connectivity_failure), auto_complete_messages_from_client_,
-      abort_on_undrained_messages_);
+      Ref(), server, std::move(on_connectivity_failure),
+      auto_complete_messages_from_client_, abort_on_undrained_messages_);
   entry = transport->Ref();
   return transport;
 }

test/core/xds/xds_transport_fake.h

@@ -156,10 +156,14 @@ class FakeXdsTransportFactory : public XdsTransportFactory {
  private:
   class FakeXdsTransport : public XdsTransport {
    public:
-    FakeXdsTransport(std::function<void(absl::Status)> on_connectivity_failure,
+    FakeXdsTransport(RefCountedPtr<FakeXdsTransportFactory> factory,
+                     const XdsBootstrap::XdsServer& server,
+                     std::function<void(absl::Status)> on_connectivity_failure,
                      bool auto_complete_messages_from_client,
                      bool abort_on_undrained_messages)
-        : auto_complete_messages_from_client_(
+        : factory_(std::move(factory)),
+          server_(server),
+          auto_complete_messages_from_client_(
               auto_complete_messages_from_client),
           abort_on_undrained_messages_(abort_on_undrained_messages),
           on_connectivity_failure_(
@@ -207,6 +211,8 @@ class FakeXdsTransportFactory : public XdsTransportFactory {
 
     void ResetBackoff() override {}
 
+    RefCountedPtr<FakeXdsTransportFactory> factory_;
+    const XdsBootstrap::XdsServer& server_;
     const bool auto_complete_messages_from_client_;
     const bool abort_on_undrained_messages_;