[karticam]

This is PR 3 of a series of PRs that add the __on_object_freed__ hook: an actor that defines the hook will be told when an object it created is freed. This PR makes the raylet record, for each plasma object, which worker created it, if that worker's actor defines the hook. Nothing reads the record yet: a later PR uses it to call the hook when the object's last copy on the node is deleted.

The complete design doc is here: https://docs.google.com/document/d/1ZJhgRmwVyKfNCR21CWf6xMcc0B6HJwSPQsfebTRvnnE/edit?tab=t.0#heading=h.7wcod1f9afpg

Changes

1. The raylet records the producer in two places, which live for different times.
   • At create: the plasma create request carries producer_worker_id into ObjectInfo. The object manager keeps it with its other per-copy info (ObjectManager::GetLocalObjectProducerWorkerId), and drops it when the in-memory copy is deleted, including when a spilled object is evicted.
   • At pin: PinObjectIDsRequest carries it into the pinned object's LocalObjectInfo (LocalObjectManager::GetProducerWorkerId). That entry is erased only when the object is really gone, so the record survives spilling and eviction. A second pin of an already pinned object keeps the first pin's producer, unless the first pin recorded none: a pin without a producer, such as a re-executed task's, can arrive before the producer's own pin.
2. Only objects the worker itself creates record it: its task returns and its Python ray.puts, at both create and pin, and only when the actor defines the hook ([core][__on_object_freed__][1/n] Detect the __on_object_freed__ hook on actor classes #67232). Everything else records Nil:
   • copies restored from spilled files (CoreWorker::CreateExisting picks the ID from its existing created_by_worker argument);
   • copies received from other nodes;
   • objects the worker stores as an owner through PutInLocalPlasmaStore (error objects and re-executed tasks' returns);
   • a task return that an earlier execution already stored (PinExistingReturnObject);
   • the owner's recovery pin and the consumer's pin of a moved object.
3. Every caller has to pass the ID. The new parameter on the plasma client's create call and on RayletClient::PinObjectIDs has no default, so each call site states the producer explicitly. Cython, Java and the C++ API callers don't change.

[gemini-code-assist]

Code Review

This pull request introduces tracking of the producer_worker_id for plasma objects created by workers whose actors define the on_object_freed hook. This ID is propagated through the core worker, plasma client, raylet, and RPC interfaces, and is stored in the local object manager. Feedback on the changes highlights critical backward-compatibility issues: inserting the new field in the middle of the PlasmaCreateRequest FlatBuffer table breaks compatibility with older binaries, and using VerifyNotNullPtr on this new field in protocol.cc will cause crashes when interacting with older clients. It is recommended to append the new field to the end of the FlatBuffer table and handle its absence gracefully by defaulting to WorkerID::Nil().

[gemini-code-assist]

In FlatBuffers, adding a new field in the middle of a table without explicit field IDs changes the sequential IDs of all subsequent fields (such as is_mutable and others). This breaks backward and forward compatibility between different versions of Ray binaries (e.g., during rolling upgrades or mixed-version clusters). To maintain compatibility, new fields should always be appended to the end of the FlatBuffer table.

[karticam]

done

[gemini-code-assist]

Using VerifyNotNullPtr on producer_worker_id will cause a crash (RAY_LOG(FATAL)) if an older client (e.g., during a rolling upgrade or mixed-version environment) sends a PlasmaCreateRequest without this newly added field. Since producer_worker_id is optional/new, we should handle its absence gracefully by defaulting to WorkerID::Nil() instead of crashing.

Suggested change

	VerifyNotNullPtr(
	message->producer_worker_id(), kProducerWorkerId, MessageType::PlasmaCreateRequest);
	object_info->producer_worker_id =
	WorkerID::FromBinary(message->producer_worker_id()->str());
	if (message->producer_worker_id() != nullptr) {
	object_info->producer_worker_id =
	WorkerID::FromBinary(message->producer_worker_id()->str());
	} else {
	object_info->producer_worker_id = WorkerID::Nil();
	}

[karticam]

The plasma client and the store always come from the same Ray build: workers connect to the plasma store inside their own node’s raylet, and Ray doesn’t support mixing versions there. The sender always sets this field, and the owner fields next to it are checked the same way, so I’ve kept the check.

[cursor]

Stale Bugbot comment from a previous run.

[cursor]

Stale Bugbot comment from a previous run.

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit e2ea787. Configure here.}