[karticam]

This is PR 5 of a series of PRs to implement move semantics for plasma objects in core.

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

After #67041, the raylet of the node where a consuming actor runs knows which of the objects it pulls are moves. This PR acts on that record when such an object arrives: the received copy is pinned as the object's new primary copy and the owner is told where the primary copy now lives. The producer's copy is not freed yet; that is in a future PR.

Changes

1. Receiving a move keeps the object alive until it is pinned. ObjectBufferPool::WriteChunk now returns whether the chunk it wrote sealed the object, and takes defer_release. With it set, the sealing chunk seals the object but keeps the plasma store reference that was taken when the object was created. The reason: the seal happens on the RPC thread, the pin happens later on the main thread, and in between a sealed copy that nobody references is just another evictable object. Holding the reference across that window is what stops the store from evicting the fresh copy before it is pinned. ReleaseObject drops the reference once the pin is done.
2. AbortCreate skips an object sealed this way (i.e. by move semantic method). This is because the store rejects aborting a sealed object, and the pull that owns the object is cancelled by the store's own "object added" notification, which in-turn triggerse AbortCreate which lands before the pin.
3. The raylet pins the move when the store reports the object as local. The pin happens in NodeManager::HandleObjectLocal. If the object has a move record, it calls PinAndReportMovedObject first. Only then does it call LeaseDependencyManager::HandleObjectLocal, which cancels the wait pull and drops the record. This order is safe for one reason. PlasmaStore::SealObjects invokes add_object_callback before Seal returns to WriteChunk, and that callback posts HandleObjectLocal to the main service. ReceiveObjectChunk posts its ReleaseObject only after WriteChunk returns. So the pin always runs before the release. The chunk ack in HandlePush is unchanged. An object without a record is handled as before.
4. NodeManager::PinAndReportMovedObject. Same steps as HandlePinObjectIDs for one object: skip if the owner is known dead. Also skip if fetch from plasma fails, if the object is gone or pending deletion. If everything is fine, it calls PinObjectsAndWaitForFree and then reports to the owner.
5. The owner learns the new primary. ObjectLocationUpdate gains a primary_moved_to_node_id field. OwnershipBasedObjectDirectory::ReportObjectPrimaryMoved sends it through the existing UpdateObjectLocationBatch. That batch already carries the ADDED update that ObjectManager::HandleObjectAdded sent for the same object, so the move report is always ordered after it. If that ADDED update is still waiting in the buffer, because a batch to the owner is in flight, the two go out as one message. On the owner, HandleUpdateObjectLocationBatch calls the existing UpdateObjectPinnedAtRaylet. From then on the consumer's node holds the primary copy. If that node dies, the object is reconstructed or reported lost, like any other primary copy. Losing the producer's node no longer loses the object.
6. ObjectManager::DebugString reports the number of move records, and NodeManager::DebugString the number of received moves pinned or not pinned. All zero on a node that never receives a move.

[gemini-code-assist]

Code Review

This pull request implements the pinning and reporting of primary copies for moved consume_once objects across nodes. It introduces deferred release mechanics in ObjectBufferPool to prevent early eviction of objects before they are pinned, updates the ObjectManager and NodeManager to handle pinning and reporting, and adds comprehensive tests. The review feedback suggests replacing a static time.sleep(1) in the new integration tests with a dynamic wait_for_condition check to avoid potential flakiness and improve test execution speed.

[gemini-code-assist]

Using time.sleep(1) to wait for the primary-moved report to reach the owner can introduce flakiness if the system is slow, or unnecessarily slow down the test suite if it is fast. Instead, we can use wait_for_condition to wait until the owner's tracked locations for ref include node_b.node_id, which guarantees that the owner has received and processed the update.

Suggested change

	# Let the primary-moved report reach the owner before the producer's node goes.
	time.sleep(1)
	# Let the primary-moved report reach the owner before the producer's node goes.
	wait_for_condition(
	lambda: node_b.node_id
	in ray.experimental.get_object_locations([ref])[ref]["node_ids"]
	)

[cursor]

Stale Bugbot comment from a previous run.

[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 71624cb. Configure here.}