Release Miles sync port claims

[TianyeGGBond]

Context

MilesModelUpdateService claims NCCL rendezvous ports in SharedStorage while building sync plans. The MILES path recorded the claim but never released it after the cache-owner transport completed, so successful syncs could leave stale MASTER_ADDR_PORT:* keys behind over a long-running job.

[JunzheJoe]

Reviewed this as part of the F1-F12 code review — supportive. Two notes before merge:

1. The failure/timeout path still leaks the claim. This PR releases on the successful-transport path (_release_port_claim fires right after run_sync_session returns — correct placement, since the port is only needed for the NCCL rendezvous during transport). But if run_sync_session raises, or the atomic unit's asyncio.wait_for cancels mid-flight (_cancel_inflight fires ray.cancel on the inflight refs but never releases the port), _release_port_claim is skipped and the claim lingers until orchestrator.kill_pipeline's delete_port_claims. Since every sync claims a fresh MASTER_ADDR_PORT:{addr}:{port} key, repeated failures still grow the claim table, and OS port reuse can collide with stale claims and burn the 8-attempt retry budget. Suggestion: keep the current post-transport release, and additionally release in the timeout/cancellation handler (or restructure as try/finally around the transport step).

2. The claim currently serves a dead path. master_addr/port is only consumed by the NCCL broadcast transport, and that path is double-guarded with NotImplementedError (service-side sync_selected_workers L177-190, miles-side _dispatch_nccl_broadcast P1-8 self-guard). cpu_serialize never touches the port. An alternative to hardening the release: skip claiming entirely until broadcast lands — which would also intersect with delete unimplemented broadcast transport path [need discussion] #24's discussion.

[TianyeGGBond]

Reopened with the hardening for your note 1 (612bdf7).

What changed

• The transport await is now wrapped in try/finally, so the release also runs when run_sync_session raises.
• The claim is threaded out of the atomic unit via a port_claim_holder list (same pattern as the existing inflight_refs), so the wait_for timeout / cancellation handlers can release it too.
• Added _release_port_claim_nowait for those handlers. One subtlety worth flagging: an awaited release is not sufficient on the cancellation path — wait_for implements the timeout by cancelling the inner task, so an await inside the finally gets cancelled with it, and CancelledError derives from BaseException, so _release_port_claim's except Exception would not catch it. The handlers therefore use a bare .remote(), which still dispatches the delete.

I drove all three exit paths against a stubbed SharedStorage — success, transport raising, and wait_for timeout. Before this commit the latter two leaked the claim; now all three release it.

On note 2 (the claim serves a dead path)

I'd rather keep the broadcast path. It's the normal, faster GPU transport and we do want it — we just don't have the setup to test the sender-side NCCL right now, so it stays behind the NotImplementedError guard as a placeholder rather than being deleted.

That makes this PR and #24 mutually exclusive:

• Keep broadcast → merge this PR; delete unimplemented broadcast transport path [need discussion] #24 gets dropped.
• Delete broadcast → merge delete unimplemented broadcast transport path [need discussion] #24; this PR becomes unnecessary.

I'm deliberately leaving #24 open rather than closing it, so the team can make the call. My vote is to keep broadcast and merge this one — the capability is necessary enough that I'd rather not delete and re-add it, even though we can't exercise it today.