[Core] Fix ObjectFetchTimedOutError

[jjyao]

Why are these changes needed?

TL;DR the bug is that for an object that's being reconstructed, we failed to set pending_creation to true so pull manager fails with ObjectFetchTimedOutError.

There are two places where we mark the object as pending_creation:

1. The creator task is executing for the first time (ReferenceCounter::UpdateSubmittedTaskReferences).
2. The creator task is re-executing to recover a lost object (ReferenceCounter::UpdateResubmittedTaskReferences).

However, if the creator task is re-executing due to failure (e.g. worker process crash, TaskManager::RetryTaskIfPossible), the return object is not marked as pending_creation, which is fine for non-streaming-generator tasks since return objects are atomically available only after the creator task finishes so they should still remain pending_creation during retry. However for streaming generator task, some return objects may be already available and used by downstream tasks when the generator task fails.

A repro sequence of events:

1. Streaming generator task A (yields 2 objects in total) starts.
2. Task A yields object a. a.pending_creation is false.
3. Task A fails, a retry A' is submitted.
4. a.pending_creation is still false.
5. Object a is lost.
6. Retrying to reconstruct a and realize there is a submitted A' so task A is not submitted again (ReferenceCounter::UpdateResubmittedTaskReferences is not called).
7. a.pending_creation is still false and will remain false (should be true) until submitted A' yields a again.

This PR fixes this issue by marking pending_creation whenever the object is lost and the creator task is running to re-create it instead of inside ReferenceCounter::UpdateResubmittedTaskReferences which might not be called.

Related issue number

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 :(

[rynewang]

As discussed offline this PR changes the semantics of pending_creation for good.