[Serve] Add broadcast API for deployment handles that broadcasts the same RPC across all live replicas of a deployment

[Crystora]

Description

Adds a broadcast() method to DeploymentHandle that calls a method on every replica of a deployment. This is useful when you need to hit all replicas - e.g. clearing a cache, reloading config, or resetting state - without relying on private internals.

Currently there's no public way to do this. Users have to dig into _router._asyncio_router._request_router to get replica references, which is fragile and breaks across versions.

Related issues

Fixes #55654

Additional information

API:

handle = serve.get_deployment_handle("MyDeployment", "default")
response = handle.broadcast("reset_cache")
results = response.results(timeout_s=10)

What changed:

• DeploymentHandle.broadcast(method_name, *args, **kwargs) - public method that bypasses load balancing and sends to all replicas directly
• DeploymentBroadcastResponse - response wrapper with results() (sync) and results_async() (async) to collect results from every replica
• Router.broadcast() - abstract method implemented across all router subclasses (SingletonThreadRouter, CurrentLoopRouter, LocalRouter)
• Added integration tests covering basic broadcast, args/kwargs, stateful mutations, async path, and single-replica edge case

Follow-up fixes (co-author: @kouroshHakha):

• Fixed coroutine single-use bug in DeploymentBroadcastResponse._fetch_replica_results_sync: in local testing mode (loop=None), if the broadcast coroutine raised an exception, calling results() a second time would crash with RuntimeError: cannot reuse already awaited coroutine. The fix caches the outcome in a concurrent.futures.Future so retries replay the cached exception.
• Added .. warning:: to broadcast() docstring noting it bypasses max_queued_requests backpressure and is intended for infrequent control-plane operations only.
• Added note to DeploymentBroadcastResponse docstring clarifying results only include replicas live at the time broadcast() was called.
• Added unit tests for the coroutine-reuse fix in TestDeploymentBroadcastResponseLocalTesting.

[gemini-code-assist]

Code Review

This pull request introduces a broadcast() method to DeploymentHandle, allowing a method to be called on all replicas of a deployment. This is a valuable feature for operations like cache clearing or state synchronization. However, the current implementation introduces significant security and stability concerns. Specifically, DeploymentBroadcastResponse.results() can block the asyncio event loop and may crash when used with CurrentLoopRouter. More critically, the broadcast() implementation in the router bypasses max_queued_requests and replica-side rejection, posing a risk of Denial of Service by allowing an attacker to overwhelm all replicas. Additionally, the broadcast implementation in the local testing router is blocking, which is inconsistent with its asynchronous interface. The main AsyncioRouter implementation of broadcast also fails to resolve arguments from other handle calls and does not propagate the tracing context, which are significant omissions.

[Crystora]

@kouroshHakha @bveeramani could you please review this PR? Welcome to any feedback.
thanks

[bveeramani]

Hey @bittoby , thanks for the PR! I don't have context on the Serve codebase so I can't review this.

Will defer to the Serve code owners

[kouroshHakha]

deferring to @abrarsheikh for review

[Crystora]

@abrarsheikh could you please review this PR? I'd appreciate any feedback

[harshit-anyscale]

@bittoby can you please fix the lint issues and test coverage failures?

[Crystora]

Update! Please review again @eicherseiji
thank you

[eicherseiji]

It looks good to me, but would like @abrarsheikh's feedback as well. @bittoby please fix lint.

[kouroshHakha]

@cursoragent can you update the serve docs to reflect this feature?

[Crystora]

Hi, @kouroshHakha what else should I need to update? It took long time since I submitted this PR. would you please merge this PR? thanks

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit a6952c7. Configure here.}

[cursor]

Timeout ignored during local-testing-mode initial fetch

Low Severity

In _fetch_replica_results_sync, when self._loop is None (local testing mode), tmp_loop.run_until_complete(self._coro) blocks indefinitely without honoring timeout_s. The timeout is only applied to self._scheduled_future.result(timeout=timeout_s), but by that point the future is already resolved. This means results(timeout_s=...) can hang forever in local testing if the broadcast coroutine itself blocks.

 

^{Reviewed by Cursor Bugbot for commit a6952c7. Configure here.}

[Crystora]

Thanks @kouroshHakha 👍