[Core] The streaming generator will mark the inplasma object that is already ready as failed after the task fails.

What happened + What you expected to happen

This is an issue that is triggered under a specific sequence of events:

1. The driver is located on node A, and the actor is located node on B. The driver retrieves data through the streaming generator.
2. The streaming generator completes for the first time. However, before the data is consumed, node B fails. The entire task is retried via ray.get. At this point, the data we get from the plasma store on node A includes the last piece of data, which confirms that all data has been successfully reconstructed.
3. The actor is killed by ray.kill.
4. When consuming all the data from the streaming generator, it is found that the data that had already been reconstructed has turned into an exception (due to the actor being killed, resulting in data loss).

This is because ray.kill triggers a function that marks all objects in the plasma store as failed. However, in reality, the plasma store on node B actually contains the complete versions of the data.

Versions / Dependencies

ray master, commit: 385e000b8d10de6d1ccc0111621cb1bce6e74f6a

Reproduction script

copy this case in test-file python/ray/tests/test_streaming_generator_4.py:

def test_streaming_generator_failed_and_try_get_yield_object(ray_start_cluster):
    cluster = ray_start_cluster
    cluster.add_node(
        num_cpus=1,
        resources={"head": 1},
        _system_config=RECONSTRUCTION_CONFIG,
        enable_object_reconstruction=True,
    )
    ray.init(address=cluster.address)

    worker = cluster.add_node(
        num_cpus=1,
        resources={"worker": 1},
    )

    @ray.remote
    class Actor:
        def __init__(self):
            self._no_wait = False

        def set_no_wait(self):
            self._no_wait = True
            return

        def stream(self):
            for _ in range(3):
                yield True
                # 100KB, in plasma
                yield b"0" * (1024 * 100 + 1)

            yield False
            if self._no_wait:
                return

            # hang in here, wait for ray.kill
            time.sleep(3600)

    @ray.remote
    def tigger_task_after_10s():
        time.sleep(10)

    @ray.remote
    def fetch_data(_, *args):
        for arg in args:
            print(len(arg))

    a = Actor.options(
        resources={"worker": 0.1}, max_restarts=-1, max_task_retries=-1
    ).remote()

    ray.get(a.set_no_wait.remote())

    gen = a.stream.remote()
    refs = []
    while True:
        has_next = next(gen)
        if not ray.get(has_next):
            break

        ref = next(gen)
        refs.append(ref)

    fetch_result = fetch_data.options(resources={"head": 0.1}).remote(
        # make sure mark all refs failed.
        tigger_task_after_10s.options(resources={"head": 0.1}).remote(),
        *refs,
    )

    cluster.remove_node(worker, allow_graceful=False)

    worker = cluster.add_node(
        num_cpus=1,
        resources={"worker": 1},
    )

    # make sure all refs ready!
    ray.get(refs[-1])

    ray.kill(a, no_restart=True)

    ray.get(fetch_result)

Issue Severity

None