fix: end transfer-lock leaks — QPN collision ownership, RAII release guard, session worker leak, fetch chunk clamp

[xiaguan]

Incident

On a 4-node H200 production cluster (v0.23.2), two holder nodes logged a steady stream of Transfer lock expired warnings (~270 sessions per holder over 11h, still growing), each session pinning up to ~1,400 blocks for the full 120s lock timeout before GC reclaimed them. Holder-side RPC counters showed the smoking gun: query_blocks_for_transfer ok=2235 vs release_transfer_lock ok=1961 — the release deficit matched the cumulative timeout count almost exactly, while the requesters logged zero fetch errors. The releases were never sent, not failing.

Root cause chain (all confirmed by logs / stderr)

1. Cross-peer QPN collision. PerNicState kept sessions and remote_memory in HashMaps keyed by the remote first QPN. QPNs are only unique within one remote HCA — peers booted from the same image at the same time allocate near-identical QPNs, so two peers' keys collide. The second peer's handshake silently overwrote the first peer's entries, so transfers to the first peer ran against the wrong remote-memory snapshot → remote memory not found in handshake snapshot.
2. Cross-peer destruction → permanent wedge. That error invalidated peer 1's connection, and cleanup by the colliding QPN key destroyed peer 2's session vec while peer 2's addr_connections entry survived. Every later fetch toward peer 2 hit .expect("session vec must exist for established connection") — a panic on v0.23.2 — and since the connection still looked Existing, it never re-handshook.
3. Panic → silent lock leak. Each panic killed the detached prefetch task (panics go to stderr only, none of the usual WARN patterns), so ReleaseTransferLock was never sent and the holder pinned the blocks until the 120s GC. Requester stderr showed 340 / 311 panic groups, each aligned ~122s before a holder-side lock timeout.

Fixes

• a9aa74e (root cause) — AddrConnection now owns its per-NIC state (ConnNic { sessions, remote_memory, rr_counter }). The QPN-keyed global maps are deleted, so the collision class is unrepresentable; invalidate_connection is one atomic remove, and a half-removed connection can no longer be observed. Also fixes the mirror wedge on the accept side: rdma_accept_handshake now aborts the handshake when complete_handshake fails instead of leaving the client stuck in connecting forever.
• 7788555 (defense in depth) — requester-side RAII TransferLockGuard: the session is released on every exit — completion, handled error, panic unwind, or the future being dropped. The drop path logs a WARN with session/remote/req_id so any remaining abort path is visible in production. Covered by a loopback tonic test asserting exactly-once release across all three exits.
• 3cea9bf — session worker threads captured Arc<RcSession> while the session owned the only command Sender, so recv() could never disconnect: every invalidated/aborted connection leaked its worker thread, QP, and CQ forever. The worker now holds a Weak and upgrades per batch.
• 562d521 — fix(core): reduce RDMA prefetch reclaim amplification #387 capped fetch staging chunks at 256 MiB but also made every fetch allocate at least that much, which fails outright on pools smaller than the cap. Found by the p2p_rdma IT (16 MiB pool), which fix(core): reduce RDMA prefetch reclaim amplification #387 had broken — confirmed by A/B against master on real hardware. Chunks are now sized min(remaining bytes on that NUMA, 256 MiB), keeping the amplification cap for large fetches.
• bb480ef / 9fa2830 — split TransferLockGuard into its own module (file back under the 1k-line limit); log real QPNs across the worker lifecycle.

Breaking metric change

pegaflow_rdma_qps previously reported map entries (peers × NICs); it now reports actual QP count, i.e. scaled up by qps_per_peer, matching its own description ("Active RC queue pairs") and the cpu_bench assertion. Expect a step in dashboards.

Validation

• p2p_rdma IT on an H200 node with 400G IB (PEGAFLOW_IB_DEVICE=mlx5_gpu*): fails on master (chunk allocation error), passes on this branch (4/4 blocks RDMA-fetched, data verified, lock released, all session workers exit cleanly on teardown — previously they leaked).
• cargo clippy --workspace --all-targets clean; 125 pegaflow-core + 17 pegaflow-transfer tests pass; pre-commit hooks (incl. cargo test --release) passed on every commit.
• Two adversarial review rounds; final verdict approve.

Follow-ups (pre-existing, out of scope, issues to be filed)

1. A local allocation failure during fetch tears down a healthy RDMA connection (invalidate + full re-handshake loop under pool pressure); string errors can't distinguish connection-broken from local-OOM.
2. rdma_accept_handshake has an unreachable! reachable under simultaneous mutual handshake (invalidate → get_or_prepare window); should return an error instead.
3. Transfer-timeout path returns pinned staging buffers to the pool while in-flight READ WQEs may still DMA into them; same family: QP destroy with outstanding WQEs after a completion error.

🤖 Generated with Claude Code

[GentleCold]

I found a small race in the new Weak<RcSession> worker path:

A: transfer_batch_async() successfully sends command into cmd_rx
B: a concurrent transfer fails and invalidate_connection() drops the last Arc<RcSession>
worker: receives A's command, but Weak::upgrade() returns None
worker: exits and drops A's command/done_tx without executing the batch

This does not leave A permanently successful: its receiver will eventually observe channel closed, and the fetch error path should still release the transfer lock. However, send() succeeding only means that A was enqueued, not that it will execute, and the current failure is reported as a generic channel-closed error.

Could the None branch notify the current command's done_tx with an explicit TransferError such as session invalidated before execution before exiting? That would avoid silently dropping an accepted command and make the invalidation race observable to the caller. This looks like a reliability/diagnostics fix rather than a data-corruption blocker.

[xiaguan]

Good catch — fixed in cd93e59. The upgrade() == None branch now replies session invalidated before batch execution to the dequeued command's done_tx, and also drains everything still queued (try_recv loop) with the same explicit error before exiting, since those commands were equally accepted-but-doomed. The generic channel-closed error is left only for commands enqueued after the worker is already gone, where send() itself fails.

Re-ran the p2p_rdma IT on the H200 + 400G IB node at this commit: passes (4/4 blocks fetched, all session workers exit cleanly).

[GentleCold]

LGTM