feat(umbp): cache remote fetches into local DRAM (dual-scheme)

[inkcherry]

Summary

cache_remote_fetches is already declared in UMBPDistributedConfig, exposed in pybind, printed in the client config dump, and documented (design-master-control-plane.md + the UMBP_CACHE_REMOTE_FETCHES env var) — but it was never implemented: a remote fetch read the peer's block into the caller and dropped it, keeping no local copy. This PR implements the documented behavior.

When a node fetches a KV block from a remote peer's L3, the block is now written back into that node's local L3 DRAM tier and re-published to the master, so subsequent reads of the same key are served locally instead of paying another cross-node RDMA round-trip.

Motivation

In multi-node UMBP serving (e.g. sglang HiCache L3 with 2 prefill nodes sharing prefixes), two nodes repeatedly read the same shared-prefix KV blocks. Without re-caching, every repeat read of a peer-owned block is a full cross-node RDMA fetch and keeps load on the owning node. Re-caching turns "remote every time" into "remote once, then local" — lowering TTFT tails and raising throughput. (sglang's own L2 only holds the block while hot in the small host cache and does not create a durable local L3 replica, so the large L3 tier stays empty for peer-owned keys without this feature.)

What changed (additive; no proto/API break)

• PoolClient (distributed/pool_client.cpp): on a successful remote Get, if cache_remote_fetches and the admission gate passes, enqueue the block for a local install — MaybeReCacheAfterRemote → async worker → ExecuteLocalPut(DRAM) → Commit → KvEvent::ADD publish. Wired into the Get/BatchGet finalize path.
• Admission gate: CacheRemoteAdmission {SIZE (default), NEVER, ALWAYS} + admission_max_block_bytes (16 MB) on UMBPDistributedConfig / PoolClientConfig.
• RouteGet (routing/route_get_strategy.cpp): prefer a best-tier replica whose node_id == requester (self-local) before random pick, so a re-cached block is actually read locally on the requester's next request.
• pybind: expose the new enum + fields.

Design notes

• Async off-path install: re-cache runs on a background worker, never on the Get return path. A synchronous first version regressed tail latency badly under concurrency (inline Allocate+memcpy+Commit blocked concurrent Gets: avg TTFT +112%, p90 +312%); the async design eliminates that.
• Worker started in Init (when enabled), drained/joined in Shutdown before the allocator is torn down. Re-cache is idempotent.

Performance — 2P1D sglang HiCache E2E, on vs off

2 prefill nodes share the UMBP L3 prefix cache; ON vs OFF differ ONLY in cache_remote_fetches. Model DeepSeek-V3-5layer, 8xMI300X/node, mori RoCE transfer, real forward. Fresh umbp_master per arm (round_0 cache-hit = 0.0, contamination-free).

This PR (dual-scheme main), two clean pairs:

metric	pair A (OFF→ON)	pair B (OFF→ON)
throughput	+8.9%	+4.4%
p99 TTFT	-76.1%	-22.6%
p90 TTFT	-92.5% (*)	-0.3%
avg TTFT	-74.8% (*)	+0.4%
cache hit	-6.0%	-4.7%

(*) pair A's OFF arm hit a tail spike; pair B's OFF baseline was already tail-clean, so its avg/p90 are flat — but ON never regresses. Robust signals (throughput up, p99 down) agree across both pairs.

Independent N=3 confirmation (same mechanism, earlier UMBP lineage, alternated order):

metric	OFF (mean±sd)	ON (mean±sd)	delta
p90 TTFT	2.422 ± 1.803	1.368 ± 1.035	-43.5%
avg TTFT	0.669 ± 0.215	0.578 ± 0.146	-13.6%
throughput	8.855 ± 0.400	9.161 ± 0.058	+3.5%
cache hit	0.812	0.804	~flat

ON p90 TTFT was lower in all 3 reps.

Note: cache-hit rate is ~flat by design — the feature accelerates the fetch path (remote→local), it does not change which prefixes hit; the value shows up in latency/throughput.

Reproduction

Consumed via sglang's HiCache UMBP backend, so E2E repro needs the companion sglang UMBP stack (see sgl-project/sglang#25377).

3 nodes — P1 = prefill + umbp_master, P2 = prefill, D = decode + router. RoCE env on every node (RoCEv2 GID index; pin one NIC == the node's UMBP address):

export MORI_IB_GID_INDEX=3 MORI_RDMA_DEVICES=mlx5_0 MORI_SOCKET_IFNAME=eth0

P1 (prefill + auto-start umbp_master):

UMBP_MASTER_ADDRESS=$P1:50051 UMBP_NODE_ADDRESS=$P1 UMBP_MASTER_AUTO_START=true \
UMBP_MASTER_BIN=<mori>/build/src/umbp/umbp_master \
UMBP_IO_ENGINE_PORT=18080 UMBP_PEER_SERVICE_PORT=19080 \
UMBP_CACHE_REMOTE_FETCHES=true UMBP_DRAM_BYTES=$((8*1024**3)) UMBP_SSD_BYTES=0 \
ENABLE_HICACHE=true ENABLE_UMBP=true HICACHE_SIZE=2 DISAGG_TRANSFER_BACKEND=mori \
MODEL_PATH=<DeepSeek-V3-5layer> bash run_pd_disagg_bench_dp8ep8.sh --role prefill

P2 (prefill, joins P1's master): same as P1 but UMBP_NODE_ADDRESS=$P2, UMBP_MASTER_AUTO_START=false, and drop UMBP_MASTER_BIN.

D (decode + router, drives the benchmark):

PREFILL_URLS="http://$P1:30000 http://$P2:30000" ENABLE_ROUTER=true \
DISAGG_TRANSFER_BACKEND=mori NUM_ROUNDS=12 NUM_CLIENTS=64 MAX_PARALLEL=32 SEED=42 \
MODEL_PATH=<DeepSeek-V3-5layer> bash run_pd_disagg_bench_dp8ep8.sh --role decode

OFF arm: rerun all three with UMBP_CACHE_REMOTE_FETCHES=false and a fresh umbp_master (kill it between arms so L3 starts cold; verify round_0 hit = 0.0). Per-request metrics land in the decode node's performance_metrics.jsonl.

Backward compatibility

Additive only. cache_remote_fetches default preserved; when off, behavior is identical to before. No proto changes; PublishLocalBlock is an existing RPC.

Testing

• mori builds with the change (pybinds + umbp_master).
• ON/OFF toggled per-arm (verified in the client config dump); cross-node remote fetch confirmed on the 2nd prefill's UMBP BatchGet logs; re-cache is idempotent.

[isytwu]

LGTM