[serve] Invalidate replica queue-length cache on gRPC request failures

[rohitsudhakar1]

Why are these changes needed?

When a DeploymentHandle uses gRPC transport (_by_reference=False), the AsyncioRouter's request-completion done-callback receives the raw grpc.aio.Call object, whereas the actor transport invokes the same callback with the request's result or a RayError.

Because of this mismatch, Router._process_finished_request can never detect a failed replica on the gRPC path:

• _get_actor_died_error(result) returns None (the result is a grpc.aio.Call, not an ActorDiedError/RayTaskError).
• isinstance(result, ActorUnavailableError) is False.

Both branches are skipped silently, so the replica's queue-length cache entry is not invalidated. After a gRPC failure, power-of-two-choices routing keeps selecting the dead replica until either the next request's rejection path invalidates the cache, or the controller's long-poll pushes a new replica set. The actor path (_by_reference=True, default) is unaffected.

Fix

Normalize a failed gRPC call (StatusCode.UNAVAILABLE) into ActorUnavailableError inside gRPCReplicaResult.add_done_callback, matching the actor-transport callback contract. This reuses the existing translation already used on the data path (_process_grpc_response / get_rejection_response), so the router's existing logic invalidates the cache and reroutes with no router changes.

CANCELLED is intentionally left untouched in the done-callback path, since there it is most often a client-initiated cancellation rather than a replica failure (turning it into a retryable error would invalidate the cache spuriously).

Related issue number

Closes #63261

Checks

• I've signed off every commit (-s) ... (will add DCO sign-off if required by CI)
• I've added unit tests covering the UNAVAILABLE, OK, and CANCELLED outcomes (python/ray/serve/tests/unit/test_grpc_replica_result.py).
• black==22.10.0 and ruff==0.8.4 (repo-pinned) pass on the changed files.

[gemini-code-assist]

Code Review

This pull request introduces normalization of failed gRPC calls into ActorUnavailableError within gRPCReplicaResult.add_done_callback to align with the actor-transport contract and prevent dead replicas from being silently ignored. It also adds corresponding unit tests to verify this behavior. The review feedback suggests optimizing the callback execution by avoiding run_coroutine_threadsafe when already running on the target event loop, and replacing the polling loop in the unit tests with an asyncio.Event to prevent flakiness.

[gemini-code-assist]

When _on_done is called, it is almost always executed on the same event loop thread (self._grpc_call_loop). Using run_coroutine_threadsafe in this case introduces unnecessary overhead (such as thread-safe queues and locking). We can optimize this by checking if we are already running on the target event loop and using create_task directly, falling back to run_coroutine_threadsafe only if called from a different thread or during shutdown.

            try:
                current_loop = asyncio.get_running_loop()
            except RuntimeError:
                current_loop = None

            if current_loop is self._grpc_call_loop and current_loop.is_running():
                current_loop.create_task(_invoke())
            else:
                coro = _invoke()
                try:
                    run_coroutine_threadsafe(coro, self._grpc_call_loop)
                except RuntimeError:
                    coro.close()
                    callback(call)

[gemini-code-assist]

Using a polling loop with asyncio.sleep(0.01) to wait for the callback can be flaky and inefficient. We can make this test completely deterministic and faster by using an asyncio.Event to wait for the callback to be invoked.

        event = asyncio.Event()
        received = []

        def callback(r):
            received.append(r)
            event.set()

        result.add_done_callback(callback)
        fake_call.complete()

        try:
            await asyncio.wait_for(event.wait(), timeout=1.0)
        except asyncio.TimeoutError:
            pytest.fail("done-callback was never invoked")

[rohitsudhakar1]

Thanks for the review — addressed both points in aba9c20:

1. Avoid run_coroutine_threadsafe on the hot path: since grpc.aio fires done-callbacks on the call's own loop, _on_done now uses loop.create_task(...) when already running on self._grpc_call_loop, and only falls back to run_coroutine_threadsafe when invoked from another thread or during shutdown.
2. Deterministic test: replaced the asyncio.sleep polling loop with an asyncio.Event + asyncio.wait_for.