[core][move semantics][7/n] Owner frees the previous primary copy after a move

[karticam]

This is PR 7 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 #67093 and #67133, the consumer's node pins a moved object as its new primary copy and reports it to the owner, but the producer's copy stays pinned until the object goes out of scope, so a move still keeps two copies. With this PR the owner frees the previous primary's copy when it handles that report, so a move ends with one copy.

Background

• The consumer's raylet reports the new primary in ObjectLocationUpdate.primary_moved_to_node_id ([core][move semantics][5/n] Raylet pins a pulled consume_once object as the primary copy and reports it to the owner #67093). The owner handles it in CoreWorker::HandleUpdateObjectLocationBatch.
• CoreWorker::FreeObjectOnNodesAsync frees an object's copies on given nodes through batched FreeLocalObjects RPCs. The owner already uses it when an object goes out of scope.
• When the primary's node dies, the owner's recovery first tries to pin an existing copy at one of the object's locations, and falls back to lineage reconstruction.
• The owner keeps one spill location per object (spilled_url, spilled_node_id), written by every spill report it receives.

Changes

1. The owner frees the previous primary's copy. The new ReferenceCounter::UpdateObjectPrimaryMoved records the new primary and returns the previous primary's node, and HandleUpdateObjectLocationBatch frees the copy there. Nothing is freed when:

   • there is no previous primary: recovery is in flight;
   • the previous primary is the new node: the owner handles the same update twice, which happens when the raylet resends a location batch after a lost reply;
   • the new node is already dead: the previous primary was never freed, so it still pins the object and stays the primary.

2. Recovery skips a copy that is being freed. If the consumer's node dies while the free is on its way to the producer's node, recovery could pin the producer's copy just before the free deletes it. The owner would then record a primary that no longer exists, and the object would be lost without an error. Each Reference now has locations_being_freed, and the new GetObjectLocationsForRecovery returns the locations minus those nodes. Only recovery's lookup uses it; every other reader still sees all locations. A node leaves the set when it reports the copy removed or when it dies, and the set is cleared when the object goes out of scope. The batched free has no per-object reply and fails only when the node is dead, so node death also covers a failed free.

3. Spill reports of a moved object. For a while both the producer's and the consumer's nodes pin a moved object, and both may spill it, but the owner keeps one spill location. A late spill report from the producer's node would point the owner at a file the free deletes, and a later death of that node would then reset the consumer's healthy primary. So:

   • HandleObjectSpilled takes the reporting node, which the owner knows from the batch even for a spill to external storage, and records a moved object's spill only from the node that holds its primary copy. While recovery is in flight, it ignores only a node whose copy is being freed.
   • The primary-moved update is handled before a spill report in the same update, since the consumer's node can spill the object before its update leaves.
   • When the previous primary is freed, its recorded spill is dropped, since the free deletes the spill file too.

   Objects that are never moved are unaffected.

Scenarios

P is the producer's node and C the consumer's node; M is the owner handling the primary-moved update. Each row ends with exactly one primary copy, or with lineage reconstruction where the row says so. Python tests are in python/ray/tests/test_consume_once_cross_node.py.

#	Scenario	What happens	Test
1	Plain move, every owner placement	P's copy is freed and its memory released; C is the only location	test_cross_node_move_pins_consumer_copy, test_move_frees_producer_memory, test_topology_*, and every #67133 test
2	c.remote(r, r), two args, or the same update handled twice	One free per object, never of C's copy	test_cross_node_move_pins_once_per_object, test_topology_repeated_and_two_args; core_worker_test.PrimaryMovedFreesPreviousPrimaryCopy
3	P's node dies after the move	The object survives on C	test_*_move_survives_producer_node_death
4	C's node dies after the move	Lineage reconstruction, or ObjectLostError when the producer's task can't be retried	test_consumer_node_death_after_move
5	C's node is already dead at M	Nothing changes: P stays the primary, spilled or not	reference_counter_test.TestUpdateObjectPrimaryMovedToDeadNode
6	C's node dies while the free is on its way to P	Recovery skips P's copy and reconstructs	reference_counter_test.TestUpdateObjectPrimaryMoved, TestCopyBeingFreedClearedOnNodeDeathAndFree
7	P had spilled the object before the move	The move is served from the spill file, which the free deletes; the spill location is dropped	test_cross_node_move_of_spilled_object; reference_counter_test.TestUpdateObjectPrimaryMovedDropsSpilledCopy
8	P's spill report reaches the owner after M, or after C's	Ignored	reference_counter_test.TestMovedObjectSpillRecordedOnlyFromPrimary
9	C spills before its update leaves, so both arrive in one update	C's spill is kept and P's copy freed	core_worker_test.PrimaryMovedWithConsumerSpillKeepsConsumerSpill
10	The consumer worker is killed mid-move	Each object ends with one primary copy	test_consumer_death_mid_move_keeps_object

Ran every //src/ray/core_worker/tests/... target, node_manager_test and object_manager_test, the full package build, test_consume_once_cross_node.py (29 passed: every check that P's copy stays is now a check that it is freed, the waits for the primary-moved report now wait for that free, and test_move_frees_producer_memory and test_consumer_node_death_after_move are new), test_consume_once.py (32), and, as a check that objects that are never moved behave as before, test_object_manager (13), test_basic_2 (24), test_actor (89), test_object_spilling (26), test_reconstruction (21), all passed.

No new benchmarks: only a moved object gets a primary-moved update, and for other objects the added work is one move-state check per spill report.