Fix pull manager retry
diff --git a/python/ray/tests/test_object_manager.py b/python/ray/tests/test_object_manager.py
index c474fced448a1..b29b9caa228f9 100644
--- a/python/ray/tests/test_object_manager.py
+++ b/python/ray/tests/test_object_manager.py
@@ -254,6 +254,48 @@ def do_transfers():
     do_transfers()
 
 
+# This is a basic test to ensure that the pull request retry timer is
+# integrated properly. To test it, we create a 2 node cluster then do the
+# following:
+# (1) Fill up the driver's object store.
+# (2) Fill up the remote node's object store.
+# (3) Try to get the remote object. This should fail due to an OOM error caused
+#     by step 1.
+# (4) Allow the local object to be evicted.
+# (5) Try to get the object again. Now the retry timer should kick in and
+#     successfuly pull the remote object.
+@pytest.mark.timeout(30)
+def test_pull_request_retry(shutdown_only):
+    cluster = Cluster()
+    cluster.add_node(num_cpus=0, num_gpus=1, object_store_memory=100 * 2**20)
+    cluster.add_node(num_cpus=1, num_gpus=0, object_store_memory=100 * 2**20)
+    cluster.wait_for_nodes()
+    ray.init(address=cluster.address)
+
+    @ray.remote
+    def put():
+        return np.zeros(64 * 2**20, dtype=np.int8)
+
+    @ray.remote(num_cpus=0, num_gpus=1)
+    def driver():
+        local_ref = ray.put(np.zeros(64 * 2**20, dtype=np.int8))
+
+        remote_ref = put.remote()
+
+        ready, _ = ray.wait([remote_ref], timeout=1)
+        assert len(ready) == 0
+
+        del local_ref
+
+        # This should always complete within 10 seconds.
+        ready, _ = ray.wait([remote_ref], timeout=20)
+        assert len(ready) > 0
+
+    # Pretend the GPU node is the driver. We do this to force the placement of
+    # the driver and `put` task on different nodes.
+    ray.get(driver.remote())
+
+
 if __name__ == "__main__":
     import pytest
     import sys
diff --git a/src/ray/object_manager/object_manager.cc b/src/ray/object_manager/object_manager.cc
index 2eb641d04ef3f..3a31da864bd09 100644
--- a/src/ray/object_manager/object_manager.cc
+++ b/src/ray/object_manager/object_manager.cc
@@ -89,14 +89,7 @@ ObjectManager::ObjectManager(asio::io_service &main_service, const NodeID &self_
       static_cast<int64_t>(1L),
       static_cast<int64_t>(config_.max_bytes_in_flight / config_.object_chunk_size))));
 
-  pull_retry_timer_.async_wait([this](const boost::system::error_code &e) {
-    RAY_CHECK(!e) << "The raylet's object manager has failed unexpectedly with error: "
-                  << e
-                  << ". Please file a bug report on here: "
-                     "https://github.com/ray-project/ray/issues";
-
-    Tick();
-  });
+  pull_retry_timer_.async_wait([this](const boost::system::error_code &e) { Tick(e); });
 
   if (plasma::plasma_store_runner) {
     store_notification_ = std::make_shared<ObjectStoreNotificationManager>(main_service);
@@ -814,6 +807,16 @@ void ObjectManager::RecordMetrics() const {
   stats::ObjectManagerPullRequests().Record(pull_manager_->NumActiveRequests());
 }
 
-void ObjectManager::Tick() { pull_manager_->Tick(); }
+void ObjectManager::Tick(const boost::system::error_code &e) {
+  RAY_CHECK(!e) << "The raylet's object manager has failed unexpectedly with error: " << e
+                << ". Please file a bug report on here: "
+                   "https://github.com/ray-project/ray/issues";
+
+  pull_manager_->Tick();
+
+  auto interval = boost::posix_time::milliseconds(config_.timer_freq_ms);
+  pull_retry_timer_.expires_from_now(interval);
+  pull_retry_timer_.async_wait([this](const boost::system::error_code &e) { Tick(e); });
+}
 
 }  // namespace ray
diff --git a/src/ray/object_manager/object_manager.h b/src/ray/object_manager/object_manager.h
index 75dbb5bd4797b..ff409eb183f2c 100644
--- a/src/ray/object_manager/object_manager.h
+++ b/src/ray/object_manager/object_manager.h
@@ -284,7 +284,7 @@ class ObjectManager : public ObjectManagerInterface,
   /// Record metrics.
   void RecordMetrics() const;
 
-  void Tick();
+  void Tick(const boost::system::error_code &e);
 
  private:
   friend class TestObjectManager;
@@ -458,7 +458,6 @@ class ObjectManager : public ObjectManagerInterface,
   const RestoreSpilledObjectCallback restore_spilled_object_;
 
   /// Pull manager retry timer .
-  /* std::unique_ptr<boost::asio::deadline_timer> pull_retry_timer_; */
   boost::asio::deadline_timer pull_retry_timer_;
 
   /// Object push manager.

<!-- Thank you for your contribution! Please review https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR. If you don't have the access to it, we will shortly find a reviewer and assign them to your PR. -->

## Why are these changes needed?

#12335 introduced a bug where the object manager's global timer was only firing once.

This PR fixes that issue by reseting the timer each time it goes off.

<!-- Please give a short summary of the change and the problem this solves. -->

## Related issue number

<!-- For example: "Closes #1234" -->

## Checks

- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for https://docs.ray.io/en/master/.
- [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(
