[XdsClient] change unit tests to cover CSDS, and fix bugs discovered (g…

…rpc#38396)

I realized that before I merge grpc#38278, I should add tests for the CSDS changes.  However, the tests weren't currently covering CSDS at all, so I wrote this PR to add that.  I will merge this PR first, then update grpc#38278 to show the CSDS changes there.

Closes grpc#38396

COPYBARA_INTEGRATE_REVIEW=grpc#38396 from markdroth:xds_client_csds_tests 0508a1a
PiperOrigin-RevId: 713042964

src/core/xds/xds_client/xds_client.cc

@@ -311,7 +311,7 @@ class XdsClient::XdsChannel::AdsCall final
         resources_seen;
     uint64_t num_valid_resources = 0;
     uint64_t num_invalid_resources = 0;
-    Timestamp update_time_ = Timestamp::Now();
+    Timestamp update_time = Timestamp::Now();
     RefCountedPtr<ReadDelayHandle> read_delay_handle;
   };
   void ParseResource(size_t idx, absl::string_view type_url,
@@ -1020,16 +1020,27 @@ void XdsClient::XdsChannel::AdsCall::ParseResource(
                                                  context->read_delay_handle);
     }
     resource_state.SetNacked(context->version, decode_status.ToString(),
-                             context->update_time_);
+                             context->update_time);
     ++context->num_invalid_resources;
     return;
   }
   // Resource is valid.
   ++context->num_valid_resources;
-  // If it didn't change, ignore it.
-  if (resource_state.HasResource() &&
+  // Check if the resource has changed.
+  const bool resource_identical =
+      resource_state.HasResource() &&
       context->type->ResourcesEqual(resource_state.resource().get(),
-                                    decode_result.resource->get())) {
+                                    decode_result.resource->get());
+  // If not changed, keep using the current decoded resource object.
+  // This should avoid wasting memory, since external watchers may be
+  // holding refs to the current object.
+  if (resource_identical) decode_result.resource = resource_state.resource();
+  // Update the resource state.
+  resource_state.SetAcked(std::move(*decode_result.resource),
+                          std::string(serialized_resource), context->version,
+                          context->update_time);
+  // If the resource didn't change, inhibit watcher notifications.
+  if (resource_identical) {
     GRPC_TRACE_LOG(xds_client, INFO)
         << "[xds_client " << xds_client() << "] " << context->type_url
         << " resource " << resource_name << " identical to current, ignoring.";
@@ -1042,10 +1053,6 @@ void XdsClient::XdsChannel::AdsCall::ParseResource(
     }
     return;
   }
-  // Update the resource state.
-  resource_state.SetAcked(std::move(*decode_result.resource),
-                          std::string(serialized_resource), context->version,
-                          context->update_time_);
   // Notify watchers.
   xds_client()->NotifyWatchersOnResourceChanged(resource_state.resource(),
                                                 resource_state.watchers(),
@@ -1336,7 +1343,9 @@ void XdsClient::ResourceState::SetNacked(const std::string& version,
 
 void XdsClient::ResourceState::SetDoesNotExist() {
   resource_.reset();
+  serialized_proto_.clear();
   client_status_ = ClientResourceStatus::DOES_NOT_EXIST;
+  failed_version_.clear();
 }
 
 absl::string_view XdsClient::ResourceState::CacheStateString() const {

src/core/xds/xds_client/xds_client.h

@@ -139,8 +139,7 @@ class XdsClient : public DualRefCounted<XdsClient> {
   Mutex* mu() ABSL_LOCK_RETURNED(&mu_) { return &mu_; }
 
   // Dumps the active xDS config to the provided
-  // envoy.service.status.v3.ClientConfig message including the config status
-  // (e.g., CLIENT_REQUESTED, CLIENT_ACKED, CLIENT_NACKED).
+  // envoy.service.status.v3.ClientConfig message.
   void DumpClientConfig(std::set<std::string>* string_pool, upb_Arena* arena,
                         envoy_service_status_v3_ClientConfig* client_config)
       ABSL_EXCLUSIVE_LOCKS_REQUIRED(&mu_);

test/core/xds/BUILD

@@ -182,6 +182,7 @@ grpc_cc_test(
         "//test/core/test_util:scoped_env_var",
         "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
         "@envoy_api//envoy/service/discovery/v3:pkg_cc_proto",
+        "@envoy_api//envoy/service/status/v3:pkg_cc_proto",
     ],
 )