[Core] Fix object reconstruction hang on arguments pending creation

[jjyao]

Why are these changes needed?

Currently inside ObjectRecoveryManager::ReconstructObject we have

auto resubmitted = task_resubmitter_->ResubmitTask(task_id, &task_deps);

  if (resubmitted) {
    reference_counter_->UpdateObjectPendingCreation(object_id, true);

The object being recovered will have pending_creation set to true and it will be set to false after the resubmitted task finishes (succeed or fail) when ReferenceCounter::UpdateFinishedTaskReferences is called. However if the task is an actor task and the actor is dead, ReferenceCounter::UpdateFinishedTaskReferences will be called BEFORE ResubmitTask returns. So basically, we set pending_creation to false first and then set to true which is obviously wrong. This PR fixes this issue by setting pending_creation to true BEFORE calling task_resubmitter_->ResubmitTask(task_id, &task_deps) so that if the actor is dead, pending_creation will be set to false inside ResubmitTask().

Related issue number

Closes #47606

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• 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 added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• 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 :(

[jjyao]

cc @Catch-Bull could you review the PR when you get a chance?

[rkooo567]

looks reasonable. can we make sure some release tests that test lineage reconstruction pass before merging?

[jjyao]

Ehh, we do have code that relies on the behavior that this piece of code is executed immediately instead of being posted to the event loop:

def is_shutdown(self) -> bool:
        """Return whether the proxy actor is shutdown.

        If the actor is dead, the health check will return RayActorError.
        """
        try:
            ray.get(self._actor_handle.check_health.remote(), timeout=0)
        except RayActorError:
            # The actor is dead, so it's ready for shutdown.
            return True

        # The actor is still alive, so it's not ready for shutdown.
        return False

[rkooo567]

I think the new approach makes sense.