[core] Clarify ref counting PublishFailure fallback paths

[Yicheng-Lu-llll]

Description

It is possible that:

• The owner's ref count reaches zero while there are still active object location subscribers. The owner must PublishFailure to all current subscribers so they receive a clean error immediately, instead of hanging until the fetch timeout fires.

• A new object location subscription arrives after the owner has already removed the ref. The owner must PublishFailure to that subscriber for the same reason: surface a clean error rather than letting them wait for a location that will never come.

The claim that "ref count reaching zero does not always mean no one is still holding the ref and might subscribe to its location" is not really a bug in the ref counting protocol. This is by current design, and could (and maybe should) be improved in the future. A detailed example follows.

First, create a root actor. The root actor then starts actor A and actor B. Based on the ownership model, killing actor A won't cause actor B to die.

root actor worker -> child actor A worker
                  -> child actor B worker      (cascade kill tree)

Now an object is created via ray.put inside the root actor. The root actor passes the ref to actor A through an actor task argument (for example, actor_a.task.remote([ref])). Actor A passes the same ref to actor B in the same way, then returns immediately. Actor B stores the ref in self.ref.

At this point actor B is borrowing the ref, but the root actor doesn't yet know: actor A's task hasn't replied, so the borrower info hasn't propagated back. Suppose actor A then dies before that reply sent to the root actor. From the root actor's perspective:

• Actor A's task failed, so its submitted task ref count drops to zero.
• The root actor never received the reply from A that would have told it actor B is borrowing.
• If the root actor also drops its own local ref, the ref count reaches zero and the root actor erases the entry.

Meanwhile actor B is still alive (killing A does not cascade to B, since they are siblings, not in a parent-child cascade relationship), still holds the ref in self.ref, and may at any point call ray.get(self.ref). This subscribes actor B to the root actor's object location channel for that object.

Two possible timings:

• Actor B subscribed before the root actor erased the entry: the PublishFailure call in EraseReference notifies actor B.
• Actor B subscribes after the entry has been erased: PublishObjectLocationSnapshot finds no entry for the object and sends PublishFailure to actor B directly.

Either way, actor B receives a clean error from ray.get instead of hanging for a long time waiting for a location that will never arrive.

[gemini-code-assist]

Code Review

This pull request updates comments and log messages in src/ray/core_worker/reference_counter.cc to provide more context regarding object reference counting and subscriber handling, specifically referencing GitHub PR #63557. I have no feedback to provide as there were no review comments to evaluate.

[Yicheng-Lu-llll]

log so the user can trace the popped up error back to this log.

[Yicheng-Lu-llll]

Not adding a log here since this is a hot path.

[Sparks0219]

Great description! 🚢