[core] Fix incorrect usage of grpc streaming API in ray syncer (#58307)

There was a video object detection Ray Data workload hang reported. 

An initial investigation by @jjyao and @dayshah observed that it was due
to an actor restart and the actor creation task was being spilled to a
raylet that had an outdated resource view. This was found by looking at
the raylet state dump. This actor creation task required 1 GPU and 1
CPU, and the raylet where this actor creation task was being spilled to
had a cluster view that reported no available GPUs. However there were
many available GPUs, and all the other raylet state dumps correctly
reported this. Furthermore in the raylet logs for the oudated raylet
there was a "Failed to send a message to node: " originating from the
ray syncer. Hence an initial hypothesis was formed that the ray syncer
retry policy was not working as intended.

A follow up investigation by @edoakes and I revealed an incorrect usage
of the grpc streaming callback API.
Currently how retries works in the ray syncer on fail to send/write is:
- OnWriteDone/OnReadDone(ok = false) is called after a failed read/write
- Disconnect() (the one in *_bidi_reactor.h!) is called which flips
_disconnected to true and calls DoDisconnect()
- DoDisconnect() notifies grpc we will no longer write to the channel
via StartWritesDone() and removes the hold via RemoveHold()
- GRPC will see that the channel is idle and has no hold so will call
OnDone()
- we've overriden OnDone() to hold a cleanup_cb that contains the retry
policy that reinitializes the bidi reactor and connects to the same
server at a repeated interval of 2 seconds until it succeeds
- fault tolerance accomplished! :) 

However from logs that we added we weren't seeing OnDone() being called
after DoDisconnect() happens. From reading the grpc streaming callback
best practices here:

https://grpc.io/docs/languages/cpp/best_practices/#callback-streaming-api
it states that "The best practice is always to read until ok=false on
the client side"
From the OnDone grpc documentation:
https://grpc.github.io/grpc/cpp/classgrpc_1_1_client_bidi_reactor.html#a51529f76deeda6416ce346291577ffa9:
it states that "Notifies the application that all operations associated
with this RPC have completed and all Holds have been removed"

Since we call StartWritesDone() and removed the hold, this should notify
grpc that all operations associated with this bidi reactor are
completed. HOWEVER reads may not be finished, i.e. we have not read all
incoming data.
Consider the following scenario:
1.) We receive a bunch of resource view messages from the GCS and have
not processed all of them
2.) OnWriteDone(ok = false) is called => Disconnected() => disconnected_
= false
3.) OnReadDone(ok = true) is called however because disconnected_ = true
we early return and STOP processing any more reads as shown below:

https://github.com/ray-project/ray/blob/275a585203bef4e48c04b46b2b7778bd8265cf46/src/ray/ray_syncer/ray_syncer_bidi_reactor_base.h#L178-L180
4.) Pending reads left in queue, and prevent grpc from calling OnDone
since not all operations are done
5.) Hang, we're left in a zombie state and drop all incoming resource
view messages and don't send any resource view updates due to the
disconnected check

Hence the solution is to remove the disconnected check in OnReadDone and
simply allow all incoming data to be read.

There's a couple of interesting observations/questions remaining:
1.) The raylet with the outdated view is the local raylet to the gcs and
we're seeing read/write errors despite being on the same node
2.) From the logs I see that the gcs syncer thinks that the channel to
the raylet syncer is still available. There's no error logs on the gcs
side, its still sending messages to the raylet. Hence even though the
raylet gets the "Failed to write error: " we don't see a corresponding
error log on the GCS side.

---------

Signed-off-by: joshlee <joshlee@anyscale.com>

Author: Sparks0219

src/ray/ray_syncer/ray_syncer_bidi_reactor.h

@@ -56,14 +56,14 @@ using ray::rpc::syncer::ResourceViewSyncMessage;
 ///
 ///
 /// For the client side:
-/// +------------+      +-------------+       +------------+  gRPC error or disconnected   +--------+
-/// | StartCall  | ---> |  StartRead  | <---> | OnReadDone | ----------------------------> | OnDone |
-/// +------------+      +-------------+       +------------+                               +--------+
-///   |                                                                                        ^
-///   |                                                                                        |
-///   v                                                                                        |
-/// +------------+      +-------------+  gRPC error or disconnected                            |
-/// | StartWrite | <--> | OnWriteDone | -------------------------------------------------------+
+/// +------------+      +-------------+       +------------+  gRPC error or ALL incoming data read   +--------+
+/// | StartCall  | ---> |  StartRead  | <---> | OnReadDone | --------------------------------------> | OnDone |
+/// +------------+      +-------------+       +------------+                                         +--------+
+///   |                                                                                                   ^
+///   |                                                                                                   |
+///   v                                                                                                   |
+/// +------------+      +-------------+  gRPC error or disconnected                                       |
+/// | StartWrite | <--> | OnWriteDone | ------------------------------------------------------------------+
 /// +------------+      +-------------+
 // clang-format on
 class RaySyncerBidiReactor {

src/ray/ray_syncer/ray_syncer_bidi_reactor_base.h

@@ -171,14 +171,13 @@ class RaySyncerBidiReactorBase : public RaySyncerBidiReactor, public T {
 
   void OnReadDone(bool ok) override {
     io_context_.dispatch(
-        [this,
-         ok,
-         disconnected = IsDisconnected(),
-         msg = std::move(receiving_message_)]() mutable {
-          if (*disconnected) {
-            return;
-          }
-
+        [this, ok, msg = std::move(receiving_message_)]() mutable {
+          // NOTE: According to the grpc callback streaming api best practices 3.)
+          // https://grpc.io/docs/languages/cpp/best_practices/#callback-streaming-api
+          // The client must read all incoming data i.e. until OnReadDone(ok = false)
+          // happens for OnDone to be called. Hence even if disconnected_ is true, we
+          // still need to allow OnReadDone to repeatedly execute until StartReadData has
+          // consumed all the data for OnDone to be called.
           if (!ok) {
             RAY_LOG_EVERY_MS(INFO, 1000) << "Failed to read a message from node: "
                                          << NodeID::FromBinary(GetRemoteNodeID());