[YutaoMa]

Motivation

The per-cluster CDS watch (ClusterConnector) and EDS diff loop (EndpointManager) tasks outlived their channel. The CDS task held an Arc<XdsCache> that could keep the cache alive forever, and the diff loop only noticed a dropped receiver on its next send.

runtime ──owns──▶ task ──Arc──▶ XdsCache ──holds──▶ watch::Sender
                   ▲                                     │
                   └────────── waits on receiver ────────┘

(The task can't exit until the sender drops, and the sender can't drop until the task releases the cache. Since the runtime owns the task, dropping the channel doesn't break the loop.)

This leak so far manifests only when the XdsChannel is dropped, so the impact is memory growth on apps creating and discarding channels. But because we plan to address #2939 cluster removal bug soon, this leak will create a race window that the discovery task may subscribe to orphaned cache entry and wait forever, thus leaking even when channel is still alive. To simplify the upcoming cluster removal PR, I'm addressing it separately first.

Solution

• ClusterConnector aborts its task on drop and subscribes before spawning, handing the task a CacheWatch instead of the cache.
• Add TaskStream<T>, an mpsc stream that aborts its producer task on drop, and run the EDS diff loop on it.