[Core] Free unconsumed object reported for deleted generator

[Kunchd]

Description

In 2.56 raylet subscribed to object owners to listen to when the objects should be evicted. However, #63181 removed this system in favor of sending free object requests to specifically the nodes that hold them instead of broadcasting to all nodes.

This change has caused a regression in the following code snippet:

@ray.remote(
        num_cpus=1,
        _generator_backpressure_num_objects=1,
    )
 def gen():
        for i in range(5):
            yield np.ones(10**7, dtype=np.uint8) * i

gen_ref = gen.remote()

del gen_ref

# the back-pressured objects will remain with the worker that created 
# even though the generator has been deleted and the object will be accessible

In the snippet above, when the streaming generator gets deleted, the items that are back pressured will be produced anyways to ensure the task runs to completion properly. For version 2.56 and before, these lines are responsible for garbage collecting the back-pressured items that got created anyways. However, after the targeted free object change. The mechanism is removed, and reported unconsumed objects sticks around even if their generator ref is deleted, leaking the objects in object store.

This PR handles this case by checking if we've received an unconsumed object after generator ref has already gone out of scope. If such objects were received, we would instead free them immediately, avoiding the object leak.

Related issues

Fixes leaking generator object that are reported after generator ref goes out of scope. Introduced in #63181.

Additional information

[gemini-code-assist]

Code Review

This pull request addresses a memory leak where unconsumed objects from a deleted streaming generator are not freed. It introduces a mechanism to immediately and asynchronously free these unconsumed plasma objects on the executor's node when the caller deletes the generator. The changes include adding a FreeObjectOnNodesCallback to the TaskManager, updating HandleReportGeneratorItemReturns to trigger this callback for unconsumed items of deleted streams, and adding corresponding unit and regression tests. The review feedback highlights a misplaced comment regarding the caller_deleted logic that became confusing after the refactoring, suggesting it be moved and merged with the comment at the top of the loop.

[gemini-code-assist]

This comment explaining the caller_deleted logic is now misplaced because the code checking caller_deleted and skipping unconsumed indices was moved to the beginning of the loop (lines 1133-1138). Keeping it here right before the backpressure check is confusing and misleading. We should remove it from here and merge its explanation into the comment at the top of the loop.

  if (backpressure_threshold != -1) {

[gemini-code-assist]

We should merge the misplaced comment from the bottom of the function here to clearly explain why we only skip unconsumed indices when caller_deleted is true, and why already-consumed indices must still be handled (due to lineage reconstruction retries).

    // Unconsumed objects for a deleted generator can no longer be used.
    // They should be freed immediately. Since the batch's lowest index is
    // unconsumed and batches are contiguous, the whole reported batch is
    // unconsumed. The objects were just created by the executor, so their only
    // copy lives on the reporting worker's node.
    //
    // Once the caller has dropped the generator, it reads no further, so any
    // index it has not already consumed is unwanted; only already-consumed
    // indices reported here are lineage-reconstruction retries of
    // still-referenced returns and must still be handled.
    if (caller_deleted && !stream_it->second.IsObjectConsumed(object_index)) {