Fix bug in ray.wait (#3445)

ray.wait depends on callbacks from the GCS to decide when an object has appeared in the cluster. The raylet crashes if a callback is received for a wait request that has already completed, but this actually can happen, depending on the order of calls. More precisely:

1. Objects A and B are put in the cluster.
2. Client calls ray.wait([A, B], num_returns=1).
3. Client subscribes to locations for A and B. Locations are cached for both, so callbacks are posted for each.
4. Callback for A fires. The wait completes and the request is removed.
5. Callback for B fires. The wait request no longer exists and raylet crashes.

src/ray/object_manager/object_manager.cc

@@ -628,9 +628,13 @@ void ObjectManager::SubscribeRemainingWaitObjects(const UniqueID &wait_id) {
                                               const ObjectID &subscribe_object_id) {
             if (!client_ids.empty()) {
               auto object_id_wait_state = active_wait_requests_.find(wait_id);
-              // We never expect to handle a subscription notification for a wait that has
-              // already completed.
-              RAY_CHECK(object_id_wait_state != active_wait_requests_.end());
+              if (object_id_wait_state == active_wait_requests_.end()) {
+                // Depending on the timing of calls to the object directory, we
+                // may get a subscription notification after the wait call has
+                // already completed. If so, then don't process the
+                // notification.
+                return;
+              }
               auto &wait_state = object_id_wait_state->second;
               RAY_CHECK(wait_state.remaining.erase(subscribe_object_id));
               wait_state.found.insert(subscribe_object_id);