xdsresolver: use the refcounted utility for cluster refcounting

[ulascansenturk]

Replaces the hand-rolled atomic.Int32 in clusterInfo with grpcsync.RefCounted, as suggested in #9304.

The "decrement, and if the count hit zero run the cleanup" logic was duplicated at three sites (SelectConfig's OnCommitted, configSelector.stop, and pruneActiveClustersAndPlugins). It now lives in a single onZero callback registered when the entry is created, which lets SelectConfig's cluster and plugin branches collapse into one path.

Entries now remove themselves from activeClusters/activePlugins once their last reference is released, so pruneActiveClustersAndPlugins is gone.

Two details worth a reviewer's attention:

• The map removal is scheduled on the serializer, because the last reference is usually released by an RPC completing on an arbitrary goroutine while the active maps may only be touched from a serializer callback. It is guarded by an identity check so that a pending removal cannot evict a newer entry that has since taken the same key, and acquisition uses TryIncrement so a dead entry is replaced rather than revived.

• For clusters, the removal is queued before unsubscribe() is called, and unsubscribe() itself stays synchronous. Queueing first matters because unsubscribing makes the dependency manager push an update onto the same serializer; if the removal were queued after, that update would regenerate a service config still containing the dropped cluster (TestResolverKeepWatchOpen_ActiveRPCs catches this). Keeping unsubscribe() out of the scheduled callback matters because that callback returns early when a newer entry has taken the key, which would otherwise leak the old subscription.

References for a config selector are now taken as each cluster is recorded, rather than in a batch after all routes are built. This keeps acquisition and release symmetric, so a config selector that fails partway through releases exactly what it acquired.

TestPruneActiveClusters is replaced by two tests covering entry reuse/release and the no-revival guard.

Fixes #9304

RELEASE NOTES: none

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 83.04%. Comparing base (ec03539) to head (9c11b34).
⚠️ Report is 2 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #9318      +/-   ##
==========================================
- Coverage   83.22%   83.04%   -0.18%     
==========================================
  Files         423      423              
  Lines       35139    35228      +89     
==========================================
+ Hits        29244    29255      +11     
- Misses       4400     4451      +51     
- Partials     1495     1522      +27     

Files with missing lines	Coverage Δ
internal/xds/resolver/serviceconfig.go	87.70% <100.00%> (+0.48%)	⬆️
internal/xds/resolver/xds_resolver.go	86.59% <100.00%> (-1.00%)	⬇️

... and 24 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[eshitachandwani]

Here , if the plugin/cluster is already present , we are not incrementing the refcount, but we are decrementing the refcount in cs.stop unconditionally. I think this will cause discrepencies. We should do a TryIncrement even if the plugin is present in the map. Or we can unconditionally call r.acquireActiveClusterInfo since that checks the map and increments the refcount.
WDYT ?

[ulascansenturk]

The counts do balance here. cs.stop() ranges over cs.plugins, which is keyed by cluster name, so it decrements once per distinct plugin however many routes named it. The if !ok guard takes exactly one ref per distinct plugin, so the two match.

Calling acquireActiveClusterInfo unconditionally would increment per route while stop() still decrements once per entry, so anything named by two routes would never be released. I tried it, and TestResolverClusterSharedByMultipleRoutes (added here) times out with cluster-A still in the service config.

[eshitachandwani]

Same as plugin

[ulascansenturk]

Same as above. cs.stop() ranges over cs.clusters, so it decrements once per distinct cluster and matches the single ref taken here. TestResolverClusterSharedByMultipleRoutes covers this path too.

[eshitachandwani]

Running unsubscribe here would mean that it can run before actually removing the entry from the activeClusters map , because TrySchedule will just schedule and return, which might cause wrong update being sent.
We might want to use ScheduleAndWait. WDYT ?

[ulascansenturk]

Good catch that unsubscribe() runs before the entry is actually deleted. It stays safe because the update it provokes goes through the same serializer: the dependency manager calls r.Update(), which queues behind the removal already scheduled, so the entry is gone by the time the config is regenerated. Swapping the two makes TestResolverKeepWatchOpen_ActiveRPCs fail.

ScheduleAndWait deadlocks though. configSelector.stop() drops the last ref from inside a serializer callback, so waiting there blocks the serializer against itself. I tried it and the package times out, with the stack showing Decrement reaching ScheduleAndWait from within CallbackSerializer.run.

[eshitachandwani]

Hey @ulascansenturk, could you please reply to the comments once you've addressed them? Also, when updating the PR, please push your new changes as separate commits rather than force-pushing to squash them. Keeping the commit history intact makes it much easier for us to review what has changed. Thanks!