perf(transfer)!: run-encode QueryBlocksForTransfer addresses and coalesce RDMA READs

[xiaguan]

Follow-up to #349 (p2p_bench example), rebased onto master after its merge. Single commit.

Problem

#349's benchmark showed the v1 cross-node fetch path is bookkeeping-bound, not bandwidth-bound. For the default shape (1000 blocks × 288 slots × 2 segments = 576k segments of 4 KiB, 2250 MiB per set, two 8×H200 nodes with 8×400G IB):

stage	ms	why
query RPC	30	response carries one fixed64 per segment → ~4.6 MB on the wire
build_transfer_tasks	43	576k TransferDesc + 288k small Vecs
submit	72	576k WQE posts
rdma_wait	91	576k CQE completions (data itself moves in <5 ms)
rebuild	13	576k Segment objects
total	~254	8.66 GiB/s

Every stage is O(segments). The fix exploits one structural fact: holder segment addresses come from sequential bump allocation, so per lane (= one slot's K or V segment across the block prefix) they form an arithmetic sequence addr(i) = base + i × stride — and within a block, all slots on the same NUMA node are contiguous.

What this PR does

Wire format (engine.proto): QueryBlocksForTransferResponse replaces repeated fixed64 segment_ptrs (one per segment) with run-encoded lanes — runs_per_lane / run_base / run_stride / run_count parallel arrays. A fully bump-allocated holder collapses every lane to a single run: the response drops from ~4.6 MB to ~13 KB. Worst case (every lane fragmented per block) is ~8.6 MB, comparable to the old format. segment_ptrs only ever existed on this unmerged PR stack, so there is no compatibility concern; the field is replaced, not reserved.

Holder (pegaflow-core/src/lib.rs): the prefix scan now verifies each block against the slot template first, then feeds segment addresses into per-lane greedy run accumulators (RunTableBuilder). Two-phase per block, so a mid-block layout mismatch never leaks partial addresses into the table.

Requester (pegaflow-core/src/transfer_runs.rs::plan_transfer, new module shared by both sides):

1. Validate the table (lane count matches template, parallel arrays in sync, per-lane counts sum to block_count, no zero-size K slots, remote spans don't overflow).
2. Decode runs, bucket per NUMA node, sort by remote base.
3. Group runs whose per-block chunks tile densely (same block range, same stride, base-adjacent, grown chunk still fits the stride). This is what recovers the real layout: K and V of one slot merge first, then every slot on the same NUMA node joins, and when block chunks are back-to-back the whole NUMA region becomes one dense group.
4. One pinned-slab carve and one (coalesced) RDMA READ per dense group; non-dense groups fall back to per-block reads with a dense local mirror. The degenerate worst case is per-segment reads — exactly the old behavior.
5. Rebuild SealedBlocks arithmetically: local(b) = group_local + (remote(b) − group_remote), every pointer bounds-checked through NumaSlab::ptr_at.

READ splitting (rdma_fetch.rs): planned descs are split at MAX_RDMA_READ_BYTES = 32 MiB. Two reasons: a single WQE must stay under the HCA max message size (1 GiB on mlx5 — the first unsplit run died with IBV_WC_LOC_LEN_ERR on a 1125 MiB READ), and multiple descriptors let the engine stripe a dense region across QPs/NICs.

Trust boundary

The run table and slot template are remote input. The planner rejects inconsistent tables (wrong lane count, bad coverage, zero-count runs, zero-size K slots, span overflow); all local offset arithmetic is checked, and every materialized pointer goes through slab bounds checks. A corrupt-but-consistent table can at worst fill our own buffers with garbage — it cannot produce overlapping or out-of-bounds local writes, and it cannot construct a zero-segment block (deferred-panic vector found in review and closed).

Results

Same two-node setup as #349, --verify (every byte, all 8 ranks) passing on all runs.

shape	#349 baseline	this PR	descs	response size
4 KiB segments, 1000 blocks (2250 MiB/set)	354.86 ms / 6.19 GiB/s	75.45 ms / 29.12 GiB/s	576,000 → 72	4.6 MB → ~13 KB
64 KiB segments, 200 blocks (7200 MiB/set)	205 ms / 35 GiB/s	88.86 ms / 79.13 GiB/s	115,200 → 226	—

Stage breakdown after (4 KiB shape, median): query 6–29 ms, build 0.17 ms, submit 0.10 ms, rdma_wait 11–21 ms, rebuild ~35 ms. Rebuild is now the dominant stage (576k Segment object constructions) — that is the next PR (seal from template metadata), kept out of this one deliberately.

Testing

• 9 unit tests in transfer_runs.rs; the plan tests all run an assert_plan_replays check that re-derives every (block, lane) address through both the desc copy and the run arithmetic and asserts they agree — i.e. they test the actual correctness property, not implementation details.
• cargo test -p pegaflow-core --lib: 117 passed.
• Hardware E2E with byte-exact verification as above. A deliberately fragmented holder cannot be constructed with the current bench CLI (sequential saves, no eviction → always dense); the degraded paths are covered by unit tests only.

Review guide

• pegaflow-core/src/transfer_runs.rs is the heart (~680 lines incl. tests): RunTableBuilder (encode), plan_transfer (validate/group/layout), group_accepts (the dense-tiling merge condition — the one function where a bug corrupts data).
• rdma_fetch.rs::fetch_blocks_via_rdma + rebuild_sealed_blocks: plan execution, 32 MiB split, cursor walk.
• lib.rs::query_blocks_for_transfer: two-phase scan feeding the builder.
• service.rs: pure plumbing.

BREAKING CHANGE: QueryBlocksForTransferResponse replaces segment_ptrs with runs_per_lane/run_base/run_stride/run_count; older peers cannot interoperate.

🤖 Generated with Claude Code