xds-client: separate resource worker from ADS server task

[W4lspirit]

Motivation

The current worker owns one ADS stream directly. Federation and fallback need physical streams with independent reconnect, generation, nonce/version, backoff, and request state. Separating that architecture from routing keeps the change reviewable and lets us verify compatibility before enabling multiple servers.

Depends on #2835. Because the dependency branch exists only on the fork, this draft temporarily targets master; after #2835 lands, it will be rebased so the diff contains only this refactor.

Description

This PR separates resource/watch coordination from a dedicated task that owns the physical ADS stream lifecycle. The server task owns transport construction, reconnect backoff, request snapshots, stream generations, and wire I/O. The resource worker retains subscription aggregation, cache transitions, watcher delivery, and ACK/NACK construction.

The implementation intentionally activates only the first configured server. It does not add fallback, bootstrap federation parsing, authority routing, or xdstp behavior.

Compatibility

The same watches produce the same initial and incremental requests, ACK/NACK ordering, reconnect version and nonce behavior, watcher events, ProcessingDone flow control, and A78 connection metrics. Stream write failures enter the same closed-stream and reconnect path as receive failures.

Bounded task-channel updates are coalesced by resource type. Dirty updates and removals remain pending until they are successfully queued, and generation-tagged ACK/resume commands cannot affect a replacement stream.

Validation

• Initial request, update, ACK/NACK, and ProcessingDone tests remain unchanged.
• Added parity coverage for reconnect version/nonce state and continued watcher delivery.
• Added coverage for A78 connected/server-failure metric transitions.
• Added coverage for full command-channel dirty resend retention.
• Added coverage for stream write failure and resubscription.
• Added coverage proving only the first configured server is active.
• cargo test -p xds-client --all-features
• cargo clippy -p xds-client --all-targets --all-features -- -D warnings

[W4lspirit]

Note :
There is an already existing issue, in the way we handle send and receive

[ejona86]

I feel like we're fighting the tools quite a lot, and now we've got to have duplication of state and constantly be draining, and we've introduced more deadlocks that previously didn't exist. I feel like this is showing "we need to change something."

In Java, we have all the processing single-threaded on one thread. If you're wanting to avoid callbacks, then the next best thing seems to be to just have a lock around significant parts of the state. Do we think that would simplify things?

[ejona86]

(note: I don't claim to fully understand this PR. The commands like Resume weren't all that clear to me, but it just seemed there's way too much coordination. So either we need to make the cut between components better, or rearchitect.)

[W4lspirit]

#2836 (comment)

I feel like we're fighting the tools quite a lot

I agree, I was not sure when I should refactor the state but I guess I'll do it now.

Do we think that would simplify things?

Yes, It should make it simpler

I'll revamp the PR around this.

[ejona86]

If we want to avoid the lock, I think we could do:

1. Spawn a task that reads from/writes to the stream (this will probably become two tasks later to fix a pre-existing RPC deadlock)
   1. The task has essentially no state
   2. It reads from an unbounded channel, and writes to the stream whatever it reads. We only enqueue to that channel after there is a Transport, to avoid it growing too much. (The task would probably block waiting for the Transport, then write to the command channel saying it is ready, passing the unbounded channel.)
   3. When it reads a message from the stream it passes the message to the command channel, along with the done one-shot channel, and it blocks on done before processing further
2. Have a centralized task that processes the commands, like today

The unbounded write channel is similar to what is being done in grpc-java (writing to an RPC can infinitely queue) and grpc-go (adsStreamImpl.pendingRequests) today (and I think the other languages, too). It is possible to make it a smarter structure that is much more bounded, but I don't think we want the extra complexity today. (There's some complexity with deletions involved.)

[YutaoMa]

^ I like the single actor with unbounded buffer approach more. I think shared state with locks would cause us the same deadlocks anyway if not used properly.

[ejona86]

Even though I said I'd have no time, I had something I really didn't want to do, so I worked on this instead.

See master...ejona86:grpc-rust:xds-client-reorg and the commit message where I point out how the existing watcher channels crumble, independent of using a locks or not. We could make them unbounded to fix that, but I question the notification API. I know @dfawley, was surprised we're using channels there, and I think it is weird to have channel-per-subscription, as we are fanning out only to fan back in with StreamMap in tonic-xds.

Maybe we should just make the channels unbounded as a short-term fix. I think the system overall does have a limit on how much they would hold. Another option is maybe to make the command channel unbounded, and then await directly on the watcher channels within the command loop, stopping all command loop processing until the watchers consume from their channels. (FWIW, grpc-java is technically unbounded in both directions, but in practice it mostly runs in-line without much queuing)