[Core] CompletePendingTask indexes reply.return_objects(0) with no size check on the streaming generator path

[LuciferYang]

What happened + What you expected to happen

CompletePendingTask indexes reply.return_objects(0) three times on the streaming generator path without checking return_objects_size():

https://github.com/ray-project/ray/blob/master/src/ray/core_worker/task_manager.cc#L1426-L1440

There is a RAY_CHECK(reply.return_objects_size() == 1) further up, which asserts the very property these reads need, but it sits inside if (reply.dynamic_return_objects_size() > 0), and that field is empty for a streaming generator, so it never runs on this path:

https://github.com/ray-project/ray/blob/master/src/ray/core_worker/task_manager.cc#L1261-L1264

The first read is unconditional: it takes the generator id and runs whenever IsStreamingGenerator() && first_execution. The other two sit behind is_application_error, so those need an application error as well as an empty reply.

Replies with return_objects empty do reach this function. ActorTaskSubmitter synthesizes one for actor creation, on both the ok and the CreationTaskError path (actor_task_submitter.cc:122-140), and a worker that is already stopping answers OK with return_objects empty (task_receiver.cc:171-177). The first is stopped by the IsStreamingGenerator guard itself, since an actor creation task is never a generator. The second is not: that task can be a generator, and what keeps the reply out is the was_cancelled_before_running check in the two submitters (normal_task_submitter.cc:593, actor_task_submitter.cc:681), which this function cannot see.

An empty RepeatedPtrField has no backing array, so the read is from near address zero. protobuf's bounds check is a DCHECK, and Ray builds --compilation_mode=opt by default, so it is compiled out.

Expected: the generator id comes from a source that cannot be empty, and the remaining reads are gated on the field being present.

Versions / Dependencies

master (3281306dd0).

Reproduction script

There is no production repro, since the submitter checks above hold today. At unit level it is enough to call CompletePendingTask directly with a streaming generator spec and a reply whose return_objects is empty:

rpc::PushTaskReply reply;  // return_objects left empty
manager_.CompletePendingTask(
    spec.TaskId(), reply, caller_address, /*is_application_error=*/false);

Issue Severity

Low: nothing reaches it today. The one path whose task can be a generator is kept out by a check in the submitters, not by anything in this function.

AI assistance

AI assistance was used to investigate and write this up. The code references were verified against master.