[Store][K8s-Native][1/N] Go-library for k8s leader election (#1910)

---------

Signed-off-by: Vladislav Nosivskoy <vladnosiv@gmail.com>

.github/workflows/ci.yml

@@ -134,10 +134,10 @@ jobs:
         MASTER_PID=$!
         sleep 3
         cd mooncake-store/go
-        export LD_LIBRARY_PATH=$GITHUB_WORKSPACE/build/mooncake-asio:$GITHUB_WORKSPACE/build/mooncake-store/src:$GITHUB_WORKSPACE/build/mooncake-transfer-engine/src:$GITHUB_WORKSPACE/build/mooncake-transfer-engine/src/common/base:$GITHUB_WORKSPACE/build/mooncake-common/etcd
+        export LD_LIBRARY_PATH=$GITHUB_WORKSPACE/build/mooncake-common:$GITHUB_WORKSPACE/build/mooncake-store/src:$GITHUB_WORKSPACE/build/mooncake-transfer-engine/src:$GITHUB_WORKSPACE/build/mooncake-transfer-engine/src/common/base:$GITHUB_WORKSPACE/build/mooncake-common/etcd
         export CGO_ENABLED=1
         export CGO_CFLAGS="-I$GITHUB_WORKSPACE/mooncake-store/include -I$GITHUB_WORKSPACE/mooncake-transfer-engine/include"
-        export CGO_LDFLAGS="-L$GITHUB_WORKSPACE/build/mooncake-store/src -L$GITHUB_WORKSPACE/build/mooncake-store/src/cachelib_memory_allocator -L$GITHUB_WORKSPACE/build/mooncake-transfer-engine/src -L$GITHUB_WORKSPACE/build/mooncake-transfer-engine/src/common/base -L$GITHUB_WORKSPACE/build/mooncake-asio -L$GITHUB_WORKSPACE/build/mooncake-common/etcd -lmooncake_store -lcachelib_memory_allocator -ltransfer_engine -lbase -lasio -letcd_wrapper -lstdc++ -lnuma -lglog -lgflags -libverbs -ljsoncpp -lzstd -lcurl -luring -lasan -lm -lgcov"
+        export CGO_LDFLAGS="-L$GITHUB_WORKSPACE/build/mooncake-store/src -L$GITHUB_WORKSPACE/build/mooncake-store/src/cachelib_memory_allocator -L$GITHUB_WORKSPACE/build/mooncake-transfer-engine/src -L$GITHUB_WORKSPACE/build/mooncake-transfer-engine/src/common/base -L$GITHUB_WORKSPACE/build/mooncake-common -L$GITHUB_WORKSPACE/build/mooncake-common/etcd -lmooncake_store -lcachelib_memory_allocator -ltransfer_engine -lbase -lasio -letcd_wrapper -lstdc++ -lnuma -lglog -lgflags -libverbs -ljsoncpp -lzstd -lcurl -luring -lasan -lm -lgcov"
         # Link cudart if CUDA is available (needed for D2H staging in mooncake_store)
         if [ -d /usr/local/cuda/lib64 ]; then export CGO_LDFLAGS="$CGO_LDFLAGS -L/usr/local/cuda/lib64 -lcudart"; fi
         ASAN_OPTIONS=detect_leaks=0:verify_asan_link_order=0 MC_METADATA_SERVER=http://127.0.0.1:8080/metadata go test -v ./tests/...
@@ -746,6 +746,18 @@ jobs:
     uses: ./.github/workflows/ci_cu13.yml
     secrets: inherit
 
+  ascend-test:
+    needs: [build, check-paths]
+    if: needs.check-paths.outputs.should-run-downstream == 'true'
+    uses: ./.github/workflows/ci_ascend.yml
+    secrets: inherit
+
+  integration-test:
+    needs: [build, check-paths]
+    if: needs.check-paths.outputs.should-run-downstream == 'true'
+    uses: ./.github/workflows/integration-test.yml
+    secrets: inherit
+
   ci-gate:
     name: CI Gate
     if: always()
@@ -758,6 +770,8 @@ jobs:
       - build-docker
       - test-wheel-ubuntu
       - build-wheel-cu13
+      - ascend-test
+      - integration-test
     runs-on: ubuntu-latest
     steps:
       - name: Check required job results

.gitignore

@@ -5,6 +5,7 @@ build_ofed4
 old
 local_test
 go.sum
+!mooncake-common/etcd/go.sum
 *.so
 bin
 mod
@@ -204,4 +205,4 @@ CLAUDE.md
 _codeql_detected_source_root
 
 # CodeBuddy Memory
-.codebuddy/
+.codebuddy/

CMakeLists.txt

@@ -45,12 +45,21 @@ option(STORE_USE_REDIS "build mooncake store with redis" OFF)
 if (STORE_USE_REDIS)
   add_compile_definitions(STORE_USE_REDIS)
 endif()
+option(STORE_USE_K8S_LEASE "build mooncake store with K8s Lease leader election" OFF)
+if (STORE_USE_K8S_LEASE)
+  if (STORE_USE_ETCD)
+    message(FATAL_ERROR "STORE_USE_K8S_LEASE and STORE_USE_ETCD cannot be enabled together because both build Go c-shared HA backends.")
+  endif()
+  if (USE_ETCD AND NOT USE_ETCD_LEGACY)
+    message(FATAL_ERROR "STORE_USE_K8S_LEASE cannot be enabled with non-legacy USE_ETCD because both build Go c-shared libraries in the same process.")
+  endif()
+  add_compile_definitions(STORE_USE_K8S_LEASE)
+endif()
 
 option(STORE_USE_JEMALLOC "Use jemalloc in mooncake store master" OFF)
 
-# Define ASIO macros before adding mooncake-asio subdirectory
+# Define ASIO macros before building targets that include ASIO headers.
 add_compile_definitions(ASIO_SEPARATE_COMPILATION ASIO_DYN_LINK)
-add_subdirectory(mooncake-asio)
 
 add_subdirectory(mooncake-common)
 include_directories(mooncake-common/etcd)

dependencies.sh

@@ -23,7 +23,7 @@ NC="\033[0m" # No Color
 # Configuration
 REPO_ROOT=`pwd`
 GITHUB_PROXY=${GITHUB_PROXY:-"https://github.com"}
-GOVER=1.23.8
+GOVER=1.25.9
 
 # Function to print section headers
 print_section() {

docs/source/deployment/ssd-offload.md

@@ -88,10 +88,11 @@ store.setup(
 | `MOONCAKE_OFFLOAD_FILE_STORAGE_PATH` | `/data/file_storage` | Absolute path to the SSD storage directory |
 | `MOONCAKE_OFFLOAD_STORAGE_BACKEND_DESCRIPTOR` | `bucket_storage_backend` | Storage backend type (see below) |
 | `MOONCAKE_OFFLOAD_LOCAL_BUFFER_SIZE_BYTES` | `1342177280` (1.25 GB) | Client-side staging buffer size |
+| `MOONCAKE_OFFLOAD_SCANMETA_ITERATOR_KEYS_LIMIT` | `20000` | Max keys processed per iteration when scanning existing SSD metadata on startup |
 | `MOONCAKE_OFFLOAD_TOTAL_SIZE_LIMIT_BYTES` | `2199023255552` (2 TB) | Maximum disk usage |
 | `MOONCAKE_OFFLOAD_TOTAL_KEYS_LIMIT` | `10000000` | Maximum number of objects on disk |
 | `MOONCAKE_OFFLOAD_HEARTBEAT_INTERVAL_SECONDS` | `10` | Interval for offload heartbeat to master (seconds) |
-| `MOONCAKE_USE_URING` | `false` | Enable io_uring for async file I/O |
+| `MOONCAKE_OFFLOAD_USE_URING` | `false` | Enable io_uring for async file I/O |
 
 ### Bucket backend settings
 
@@ -101,8 +102,8 @@ Applies when `MOONCAKE_OFFLOAD_STORAGE_BACKEND_DESCRIPTOR=bucket_storage_backend
 |---|---|---|
 | `MOONCAKE_OFFLOAD_BUCKET_SIZE_LIMIT_BYTES` | `268435456` (256 MB) | Max size per bucket |
 | `MOONCAKE_OFFLOAD_BUCKET_KEYS_LIMIT` | `500` | Max keys per bucket |
-| `MOONCAKE_BUCKET_MAX_TOTAL_SIZE` | `0` | Eviction threshold in bytes. When set to `0`, the backend uses **90% of the physical disk capacity** as the quota — it does not mean unlimited. Set an explicit value to control disk usage precisely. |
-| `MOONCAKE_BUCKET_EVICTION_POLICY` | `none` | Eviction policy: `none` / `fifo` / `lru` |
+| `MOONCAKE_OFFLOAD_BUCKET_MAX_TOTAL_SIZE` | `0` | Eviction threshold in bytes. When set to `0`, the backend uses **90% of the physical disk capacity** as the quota — it does not mean unlimited. Set an explicit value to control disk usage precisely. |
+| `MOONCAKE_OFFLOAD_BUCKET_EVICTION_POLICY` | `none` | Eviction policy: `none` / `fifo` / `lru` |
 
 ---
 
@@ -127,6 +128,11 @@ Best for: general-purpose use, large-scale deployments.
 
 Stores each object in an individual file. Simple and easy to inspect, but generates many small files at scale.
 
+| Environment Variable | Default | Description |
+|---|---|---|
+| `MOONCAKE_OFFLOAD_FSDIR` | `file_per_key_dir` | Subdirectory name under `MOONCAKE_OFFLOAD_FILE_STORAGE_PATH` where objects are stored |
+| `MOONCAKE_OFFLOAD_ENABLE_EVICTION` | `true` | Enable disk eviction when the total size exceeds the quota |
+
 Best for: debugging or small-scale deployments.
 
 ### `offset_allocator_storage_backend`
@@ -143,7 +149,7 @@ Best for: high-concurrency scenarios with many small objects where restart durab
 
 ## Eviction (Bucket Backend Only)
 
-When `MOONCAKE_BUCKET_MAX_TOTAL_SIZE` is set, the backend automatically evicts buckets before writing new ones if total disk usage would exceed the limit.
+When `MOONCAKE_OFFLOAD_BUCKET_MAX_TOTAL_SIZE` is set, the backend automatically evicts buckets before writing new ones if total disk usage would exceed the limit.
 
 | Policy | Behavior |
 |--------|----------|
@@ -178,8 +184,8 @@ mooncake_master \
 ```bash
 export MOONCAKE_OFFLOAD_FILE_STORAGE_PATH=/nvme/mooncake_offload
 export MOONCAKE_OFFLOAD_STORAGE_BACKEND_DESCRIPTOR=bucket_storage_backend
-export MOONCAKE_BUCKET_MAX_TOTAL_SIZE=$((200 * 1024 * 1024 * 1024))  # 200 GB
-export MOONCAKE_BUCKET_EVICTION_POLICY=lru
+export MOONCAKE_OFFLOAD_BUCKET_MAX_TOTAL_SIZE=$((200 * 1024 * 1024 * 1024))  # 200 GB
+export MOONCAKE_OFFLOAD_BUCKET_EVICTION_POLICY=lru
 
 mooncake_client \
     --master_server_address="192.168.1.10:50051" \
@@ -250,7 +256,7 @@ mooncake_client \
 
 ### "Failed to register buffer with UringFile" warning in logs
 
-This warning appears when `MOONCAKE_USE_URING=true` and the io_uring fixed-buffer registration fails. The most common cause is that `MOONCAKE_OFFLOAD_LOCAL_BUFFER_SIZE_BYTES` exceeds the process's locked-memory limit (`RLIMIT_MEMLOCK`). io_uring requires the registered buffer to be pinned in physical memory, which counts against this limit.
+This warning appears when `MOONCAKE_OFFLOAD_USE_URING=true` and the io_uring fixed-buffer registration fails. The most common cause is that `MOONCAKE_OFFLOAD_LOCAL_BUFFER_SIZE_BYTES` exceeds the process's locked-memory limit (`RLIMIT_MEMLOCK`). io_uring requires the registered buffer to be pinned in physical memory, which counts against this limit.
 
 Check the current limit:
 

docs/source/design/ssd-offload.md

@@ -154,7 +154,7 @@ A single pre-allocated file (`kv_cache.data`) is shared by all objects. Space wi
 
 ## Eviction (BucketStorageBackend)
 
-When `MOONCAKE_BUCKET_MAX_TOTAL_SIZE` is set, the backend evicts existing buckets to make room before writing a new one. Eviction is disabled by default (`BucketEvictionPolicy::NONE`).
+When `MOONCAKE_OFFLOAD_BUCKET_MAX_TOTAL_SIZE` is set, the backend evicts existing buckets to make room before writing a new one. Eviction is disabled by default (`BucketEvictionPolicy::NONE`).
 
 ### Policies
 
@@ -195,7 +195,7 @@ This ordering guarantees:
 
 ## io_uring File I/O
 
-When `MOONCAKE_USE_URING=true`, the storage backends replace POSIX `pread`/`pwrite` calls with an io_uring-based implementation (`UringFile`). The design prioritizes eliminating inter-thread lock contention, which was the dominant latency source in the previous global-ring approach.
+When `MOONCAKE_OFFLOAD_USE_URING=true`, the storage backends replace POSIX `pread`/`pwrite` calls with an io_uring-based implementation (`UringFile`). The design prioritizes eliminating inter-thread lock contention, which was the dominant latency source in the previous global-ring approach.
 
 ### Thread-local rings (`SharedUringRing`)
 

docs/source/design/transfer-engine/efa_transport.md

@@ -81,7 +81,7 @@ make -j$(nproc)
 # Copy built modules to wheel directory
 cp mooncake-integration/engine.cpython-*.so ../mooncake-wheel/mooncake/
 cp mooncake-integration/store.cpython-*.so ../mooncake-wheel/mooncake/
-cp mooncake-asio/libasio.so ../mooncake-wheel/mooncake/
+cp mooncake-common/libasio.so ../mooncake-wheel/mooncake/
 
 # Install with pip
 pip install -e ../mooncake-wheel --no-build-isolation

docs/source/design/transfer-engine/kunpeng_ub_transport.md

@@ -99,7 +99,7 @@ make -j$(nproc)
 # Copy built modules to wheel directory
 cp mooncake-integration/engine.cpython-*.so ../mooncake-wheel/mooncake/
 cp mooncake-integration/store.cpython-*.so ../mooncake-wheel/mooncake/
-cp mooncake-asio/libasio.so ../mooncake-wheel/mooncake/
+cp mooncake-common/libasio.so ../mooncake-wheel/mooncake/
 
 # Install with pip
 pip install -e ../mooncake-wheel --no-build-isolation

docs/source/performance/ssd-offload-benchmark-results.md

@@ -141,7 +141,7 @@ mooncake_master \
 # SSD Offload Buffer: 20GB
 MOONCAKE_OFFLOAD_FILE_STORAGE_PATH="/mnt/data/file_storage" \
 MOONCAKE_OFFLOAD_LOCAL_BUFFER_SIZE_BYTES=21474836480 \
-MOONCAKE_USE_URING=1 \
+MOONCAKE_OFFLOAD_USE_URING=1 \
 mooncake_client \
     --host=127.0.0.1 \
     --global_segment_size=80GB \

mooncake-asio/CMakeLists.txt

@@ -1,50 +0,0 @@
-# Build asio as a shared library to avoid ODR violations
-# when multiple shared libraries use asio
-
-# Try to find ASIO using find_package first
-find_package(asio QUIET)
-
-if(asio_FOUND)
-    message(STATUS "Found ASIO via find_package")
-    set(ASIO_INCLUDE_DIR ${asio_INCLUDE_DIR})
-else()
-    # Fallback to find_path if find_package fails
-    find_path(ASIO_INCLUDE_DIR
-        NAMES asio.hpp
-        PATHS
-            /usr/local/include
-            /usr/include
-            ${CMAKE_INSTALL_PREFIX}/include
-        DOC "Path to ASIO headers"
-    )
-
-    if(NOT ASIO_INCLUDE_DIR)
-        message(FATAL_ERROR "ASIO not found. Please install ASIO or set ASIO_INCLUDE_DIR manually.")
-    endif()
-
-    message(STATUS "Found ASIO at: ${ASIO_INCLUDE_DIR}")
-endif()
-
-add_library(asio_shared SHARED asio_impl.cpp)
-
-target_compile_definitions(asio_shared
-    PUBLIC
-        ASIO_SEPARATE_COMPILATION
-        ASIO_DYN_LINK
-)
-
-target_include_directories(asio_shared
-    PUBLIC
-        ${ASIO_INCLUDE_DIR}
-)
-
-set_target_properties(asio_shared PROPERTIES
-    POSITION_INDEPENDENT_CODE ON
-    INSTALL_RPATH "$ORIGIN"
-    BUILD_WITH_INSTALL_RPATH TRUE
-    OUTPUT_NAME "asio"
-)
-
-target_link_libraries(asio_shared PUBLIC pthread)
-
-install(TARGETS asio_shared DESTINATION lib)

mooncake-common/CMakeLists.txt

@@ -2,6 +2,10 @@ if ((USE_ETCD AND NOT USE_ETCD_LEGACY) OR STORE_USE_ETCD)
     add_subdirectory(etcd)
 endif()
 
+if (STORE_USE_K8S_LEASE)
+    add_subdirectory(k8s-lease)
+endif()
+
 include_directories(${CMAKE_CURRENT_SOURCE_DIR}/include)
 add_subdirectory(src)
 

mooncake-common/etcd/CMakeLists.txt

@@ -3,7 +3,7 @@ add_custom_command(
     COMMAND bash -c "go mod tidy" && bash -c "go build -buildmode=c-shared -o ${CMAKE_CURRENT_BINARY_DIR}/libetcd_wrapper.so etcd_wrapper.go" && cp ${CMAKE_CURRENT_BINARY_DIR}/libetcd_wrapper.h ${CMAKE_CURRENT_SOURCE_DIR}
     WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}
     COMMENT "Building Go shared library"
-    DEPENDS etcd_wrapper.go
+    DEPENDS etcd_wrapper.go go.mod go.sum build.sh
 )
 
 set(ETCD_WRAPPER_INCLUDE ${CMAKE_CURRENT_BINARY_DIR}/libetcd_wrapper.h)
@@ -17,4 +17,4 @@ add_custom_target(
 install(
     FILES ${ETCD_WRAPPER_LIB}
     DESTINATION lib
-)
+)

mooncake-common/etcd/go.mod

@@ -1,6 +1,8 @@
 module github.com/kvcache-ai/Mooncake/mooncake-common/etcd
 
-go 1.24.0
+go 1.25.0
+
+toolchain go1.25.9
 
 require (
 	go.etcd.io/etcd/api/v3 v3.5.21

mooncake-common/etcd/go.sum

@@ -0,0 +1,108 @@
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmfM=
+github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
+github.com/coreos/go-systemd/v22 v22.3.2 h1:D9/bQk5vlXQFZ6Kwuu6zaiXJ9oTPe68++AzAJc1DzSI=
+github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
+github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
+github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
+github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
+github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
+github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
+github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+go.etcd.io/etcd/api/v3 v3.5.21 h1:A6O2/JDb3tvHhiIz3xf9nJ7REHvtEFJJ3veW3FbCnS8=
+go.etcd.io/etcd/api/v3 v3.5.21/go.mod h1:c3aH5wcvXv/9dqIw2Y810LDXJfhSYdHQ0vxmP3CCHVY=
+go.etcd.io/etcd/client/pkg/v3 v3.5.21 h1:lPBu71Y7osQmzlflM9OfeIV2JlmpBjqBNlLtcoBqUTc=
+go.etcd.io/etcd/client/pkg/v3 v3.5.21/go.mod h1:BgqT/IXPjK9NkeSDjbzwsHySX3yIle2+ndz28nVsjUs=
+go.etcd.io/etcd/client/v3 v3.5.21 h1:T6b1Ow6fNjOLOtM0xSoKNQt1ASPCLWrF9XMHcH9pEyY=
+go.etcd.io/etcd/client/v3 v3.5.21/go.mod h1:mFYy67IOqmbRf/kRUvsHixzo3iG+1OF2W2+jVIQRAnU=
+go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
+go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
+go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48=
+go.opentelemetry.io/otel v1.39.0/go.mod h1:kLlFTywNWrFyEdH0oj2xK0bFYZtHRYUdv1NklR/tgc8=
+go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0=
+go.opentelemetry.io/otel/metric v1.39.0/go.mod h1:jrZSWL33sD7bBxg1xjrqyDjnuzTUB0x1nBERXd7Ftcs=
+go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18=
+go.opentelemetry.io/otel/sdk v1.39.0/go.mod h1:vDojkC4/jsTJsE+kh+LXYQlbL8CgrEcwmt1ENZszdJE=
+go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8=
+go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew=
+go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI=
+go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA=
+go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
+go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
+go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
+go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
+go.uber.org/zap v1.17.0 h1:MTjgFu6ZLKvY6Pvaqk97GlxNBuMpV4Hy/3P6tRGlI2U=
+go.uber.org/zap v1.17.0/go.mod h1:MXVU+bhUf/A7Xi2HNOnopQOrmycQ5Ih87HtOu4q5SSo=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU=
+golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk=
+golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU=
+golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
+gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
+google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 h1:fCvbg86sFXwdrl5LgVcTEvNC+2txB5mgROGmRL5mrls=
+google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:+rXWjjaukWZun3mLfjmVnQi18E1AsFbDN9QdJ5YXLto=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 h1:gRkg/vSppuSQoDjxyiGfN4Upv/h/DQmIR10ZU8dh4Ww=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
+google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE=
+google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ=
+google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
+google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
+gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

mooncake-common/k8s-lease/CMakeLists.txt

@@ -0,0 +1,20 @@
+add_custom_command(
+    OUTPUT ${CMAKE_CURRENT_BINARY_DIR}/libk8s_lease_wrapper.so
+    COMMAND bash -c "go mod tidy" && bash -c "go build -buildmode=c-shared -o ${CMAKE_CURRENT_BINARY_DIR}/libk8s_lease_wrapper.so k8s_lease_wrapper.go" && cp ${CMAKE_CURRENT_BINARY_DIR}/libk8s_lease_wrapper.h ${CMAKE_CURRENT_SOURCE_DIR}
+    WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}
+    COMMENT "Building K8s Lease Go shared library"
+    DEPENDS k8s_lease_wrapper.go
+)
+
+set(K8S_LEASE_WRAPPER_INCLUDE ${CMAKE_CURRENT_BINARY_DIR}/libk8s_lease_wrapper.h)
+set(K8S_LEASE_WRAPPER_LIB ${CMAKE_CURRENT_BINARY_DIR}/libk8s_lease_wrapper.so)
+
+add_custom_target(
+    build_k8s_lease_wrapper
+    DEPENDS ${K8S_LEASE_WRAPPER_LIB}
+)
+
+install(
+    FILES ${K8S_LEASE_WRAPPER_LIB}
+    DESTINATION lib
+)

mooncake-common/k8s-lease/cmd/envtest-server/main.go

@@ -0,0 +1,61 @@
+// envtest-server starts a real kube-apiserver + etcd via envtest, writes the
+// KUBECONFIG path to stdout, and blocks until SIGTERM or SIGINT.  This lets
+// C++ tests launch it as a subprocess and talk to a real K8s API without a
+// full cluster.
+package main
+
+import (
+	"fmt"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"syscall"
+
+	"k8s.io/client-go/tools/clientcmd"
+	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+)
+
+func main() {
+	env := &envtest.Environment{}
+
+	cfg, err := env.Start()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "envtest start failed: %v\n", err)
+		os.Exit(1)
+	}
+
+	// Write a KUBECONFIG file that points at the envtest kube-apiserver.
+	kubeconfigPath := filepath.Join(os.TempDir(), fmt.Sprintf("envtest-kubeconfig-%d", os.Getpid()))
+	kubeconfig := clientcmdapi.NewConfig()
+	kubeconfig.Clusters["envtest"] = &clientcmdapi.Cluster{
+		Server:                   cfg.Host,
+		CertificateAuthorityData: cfg.CAData,
+	}
+	kubeconfig.AuthInfos["envtest"] = &clientcmdapi.AuthInfo{
+		ClientCertificateData: cfg.CertData,
+		ClientKeyData:         cfg.KeyData,
+	}
+	kubeconfig.Contexts["envtest"] = &clientcmdapi.Context{
+		Cluster:  "envtest",
+		AuthInfo: "envtest",
+	}
+	kubeconfig.CurrentContext = "envtest"
+
+	if err := clientcmd.WriteToFile(*kubeconfig, kubeconfigPath); err != nil {
+		fmt.Fprintf(os.Stderr, "failed to write kubeconfig: %v\n", err)
+		env.Stop()
+		os.Exit(1)
+	}
+
+	// Print the kubeconfig path — the parent process reads this from stdout.
+	fmt.Println(kubeconfigPath)
+
+	// Block until SIGTERM or SIGINT.
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
+	<-sigCh
+
+	os.Remove(kubeconfigPath)
+	env.Stop()
+}

mooncake-common/k8s-lease/go.mod

@@ -0,0 +1,60 @@
+module github.com/kvcache-ai/Mooncake/mooncake-common/k8s-lease
+
+go 1.24.0
+
+require (
+	k8s.io/api v0.34.3
+	k8s.io/apimachinery v0.34.3
+	k8s.io/client-go v0.34.3
+	k8s.io/utils v0.0.0-20251002143259-bc988d571ff4
+	sigs.k8s.io/controller-runtime v0.22.5
+)
+
+require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/emicklei/go-restful/v3 v3.12.2 // indirect
+	github.com/evanphx/json-patch/v5 v5.9.11 // indirect
+	github.com/fxamacker/cbor/v2 v2.9.0 // indirect
+	github.com/go-logr/logr v1.4.3 // indirect
+	github.com/go-openapi/jsonpointer v0.21.0 // indirect
+	github.com/go-openapi/jsonreference v0.20.2 // indirect
+	github.com/go-openapi/swag v0.23.0 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/google/gnostic-models v0.7.0 // indirect
+	github.com/google/go-cmp v0.7.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/josharian/intern v1.0.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/mailru/easyjson v0.7.7 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_golang v1.23.2 // indirect
+	github.com/prometheus/client_model v0.6.2 // indirect
+	github.com/prometheus/common v0.66.1 // indirect
+	github.com/prometheus/procfs v0.16.1 // indirect
+	github.com/spf13/pflag v1.0.9 // indirect
+	github.com/x448/float16 v0.8.4 // indirect
+	go.yaml.in/yaml/v2 v2.4.3 // indirect
+	go.yaml.in/yaml/v3 v3.0.4 // indirect
+	golang.org/x/net v0.47.0 // indirect
+	golang.org/x/oauth2 v0.30.0 // indirect
+	golang.org/x/sys v0.38.0 // indirect
+	golang.org/x/term v0.37.0 // indirect
+	golang.org/x/text v0.31.0 // indirect
+	golang.org/x/time v0.9.0 // indirect
+	google.golang.org/protobuf v1.36.8 // indirect
+	gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
+	gopkg.in/inf.v0 v0.9.1 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+	k8s.io/apiextensions-apiserver v0.34.3 // indirect
+	k8s.io/klog/v2 v2.130.1 // indirect
+	k8s.io/kube-openapi v0.0.0-20250910181357-589584f1c912 // indirect
+	sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect
+	sigs.k8s.io/randfill v1.0.0 // indirect
+	sigs.k8s.io/structured-merge-diff/v6 v6.3.2-0.20260122202528-d9cc6641c482 // indirect
+	sigs.k8s.io/yaml v1.6.0 // indirect
+)

mooncake-common/k8s-lease/k8s_lease_integration_test.go

@@ -0,0 +1,571 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+)
+
+var (
+	testEnv    *envtest.Environment
+	testConfig *rest.Config
+)
+
+type electionStateNoRelease struct {
+	cancel  context.CancelFunc
+	elected chan struct{}
+	lost    chan struct{}
+}
+
+func TestMain(m *testing.M) {
+	testEnv = &envtest.Environment{}
+
+	var err error
+	testConfig, err = testEnv.Start()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "failed to start envtest: %v\n", err)
+		os.Exit(1)
+	}
+
+	// Set up global client for the wrapper
+	client, err := kubernetes.NewForConfig(testConfig)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "failed to create clientset: %v\n", err)
+		testEnv.Stop()
+		os.Exit(1)
+	}
+	clientMutex.Lock()
+	globalClient = client
+	clientMutex.Unlock()
+
+	code := m.Run()
+
+	testEnv.Stop()
+	os.Exit(code)
+}
+
+func runElectionWithoutRelease(namespace, leaseName, identity string,
+	leaseDurationSec, renewDeadlineSec, retryPeriodSec int) (*electionStateNoRelease, error) {
+	if err := ensureClientInitialized(); err != nil {
+		return nil, err
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	state := &electionStateNoRelease{
+		cancel:  cancel,
+		elected: make(chan struct{}),
+		lost:    make(chan struct{}),
+	}
+
+	lock := &resourcelock.LeaseLock{
+		LeaseMeta: metav1.ObjectMeta{
+			Name:      leaseName,
+			Namespace: namespace,
+		},
+		Client: globalClient.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: identity,
+		},
+	}
+
+	le, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
+		Lock:            lock,
+		LeaseDuration:   time.Duration(leaseDurationSec) * time.Second,
+		RenewDeadline:   time.Duration(renewDeadlineSec) * time.Second,
+		RetryPeriod:     time.Duration(retryPeriodSec) * time.Second,
+		ReleaseOnCancel: false,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(ctx context.Context) {
+				close(state.elected)
+				<-ctx.Done()
+			},
+			OnStoppedLeading: func() {
+				close(state.lost)
+			},
+		},
+	})
+	if err != nil {
+		cancel()
+		return nil, fmt.Errorf("failed to create leader elector: %w", err)
+	}
+
+	go le.Run(ctx)
+	return state, nil
+}
+
+// TestSingleLeaderElection verifies a single candidate becomes leader.
+func TestSingleLeaderElection(t *testing.T) {
+	ns := "default"
+	lease := "single-election-test"
+	identity := "node-1:8080"
+
+	err := runElection(ns, lease, identity, 5, 4, 1)
+	if err != nil {
+		t.Fatalf("runElection failed: %v", err)
+	}
+
+	// Wait for elected
+	key := electionKey(ns, lease)
+	electionMutex.Lock()
+	state := elections[key]
+	electionMutex.Unlock()
+
+	select {
+	case <-state.elected:
+		// success
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out waiting for election")
+	}
+
+	// Verify holder via getHolder
+	holder, transitions, err := getHolder(ns, lease)
+	if err != nil {
+		t.Fatalf("getHolder failed: %v", err)
+	}
+	if holder != identity {
+		t.Errorf("expected holder %q, got %q", identity, holder)
+	}
+	// First election — transitions should be 0 or 1
+	if transitions < 0 {
+		t.Errorf("expected non-negative transitions, got %d", transitions)
+	}
+
+	// Cancel the election
+	electionMutex.Lock()
+	state = elections[key]
+	electionMutex.Unlock()
+	state.cancel()
+
+	select {
+	case <-state.lost:
+		// success
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for election loss after cancel")
+	}
+}
+
+// TestLeaderEpoch verifies leaseTransitions increments across elections.
+func TestLeaderEpoch(t *testing.T) {
+	ns := "default"
+	lease := "epoch-test"
+
+	// First election
+	err := runElection(ns, lease, "node-epoch-1:8080", 5, 4, 1)
+	if err != nil {
+		t.Fatalf("first runElection failed: %v", err)
+	}
+
+	key := electionKey(ns, lease)
+	electionMutex.Lock()
+	state1 := elections[key]
+	electionMutex.Unlock()
+
+	select {
+	case <-state1.elected:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out on first election")
+	}
+
+	_, trans1, _ := getHolder(ns, lease)
+
+	// Cancel first election and wait for loss
+	state1.cancel()
+	select {
+	case <-state1.lost:
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for first election loss")
+	}
+
+	// Wait for lease to expire / be released
+	time.Sleep(2 * time.Second)
+
+	// Second election
+	err = runElection(ns, lease, "node-epoch-2:8080", 5, 4, 1)
+	if err != nil {
+		t.Fatalf("second runElection failed: %v", err)
+	}
+
+	electionMutex.Lock()
+	state2 := elections[key]
+	electionMutex.Unlock()
+
+	select {
+	case <-state2.elected:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out on second election")
+	}
+
+	_, trans2, _ := getHolder(ns, lease)
+	if trans2 <= trans1 {
+		t.Errorf("expected transitions to increment: first=%d, second=%d", trans1, trans2)
+	}
+
+	state2.cancel()
+	select {
+	case <-state2.lost:
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for second election loss")
+	}
+}
+
+// TestSequentialLeadershipHandoff tests that a second candidate can acquire
+// leadership after the first one releases it.
+func TestSequentialLeadershipHandoff(t *testing.T) {
+	ns := "default"
+	lease := "two-candidate-test"
+
+	err1 := runElection(ns, lease, "candidate-a:8080", 5, 4, 1)
+	if err1 != nil {
+		t.Fatalf("first runElection failed: %v", err1)
+	}
+
+	key := electionKey(ns, lease)
+	electionMutex.Lock()
+	stateA := elections[key]
+	electionMutex.Unlock()
+
+	// Wait for first candidate to win
+	select {
+	case <-stateA.elected:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out waiting for first candidate")
+	}
+
+	// Verify holder is candidate-a
+	holder, _, err := getHolder(ns, lease)
+	if err != nil {
+		t.Fatalf("getHolder failed: %v", err)
+	}
+	if holder != "candidate-a:8080" {
+		t.Errorf("expected candidate-a, got %q", holder)
+	}
+
+	// Cancel candidate-a
+	stateA.cancel()
+	select {
+	case <-stateA.lost:
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for candidate-a loss")
+	}
+
+	// Wait for lease to expire
+	time.Sleep(2 * time.Second)
+
+	// Start candidate-b
+	err2 := runElection(ns, lease, "candidate-b:8080", 5, 4, 1)
+	if err2 != nil {
+		t.Fatalf("second runElection failed: %v", err2)
+	}
+
+	electionMutex.Lock()
+	stateB := elections[key]
+	electionMutex.Unlock()
+
+	select {
+	case <-stateB.elected:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out waiting for candidate-b")
+	}
+
+	holder, _, err = getHolder(ns, lease)
+	if err != nil {
+		t.Fatalf("getHolder after takeover failed: %v", err)
+	}
+	if holder != "candidate-b:8080" {
+		t.Errorf("expected candidate-b, got %q", holder)
+	}
+
+	stateB.cancel()
+	select {
+	case <-stateB.lost:
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for candidate-b loss")
+	}
+}
+
+// TestConcurrentCandidateElection starts two candidates simultaneously and
+// verifies that exactly one wins leadership.
+func TestConcurrentCandidateElection(t *testing.T) {
+	ns := "default"
+	lease := "concurrent-election-test"
+
+	type result struct {
+		identity string
+		elected  bool
+	}
+
+	candidates := []string{"candidate-a:8080", "candidate-b:8080"}
+	results := make(chan result, len(candidates))
+
+	lock := func(identity string) *resourcelock.LeaseLock {
+		return &resourcelock.LeaseLock{
+			LeaseMeta: metav1.ObjectMeta{
+				Name:      lease,
+				Namespace: ns,
+			},
+			Client: globalClient.CoordinationV1(),
+			LockConfig: resourcelock.ResourceLockConfig{
+				Identity: identity,
+			},
+		}
+	}
+
+	var wg sync.WaitGroup
+	for _, id := range candidates {
+		wg.Add(1)
+		go func(identity string) {
+			defer wg.Done()
+
+			// Short timeout: enough for one to acquire, but the loser
+			// times out before the winner's lease could expire.
+			ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
+			defer cancel()
+
+			elected := make(chan struct{})
+			le, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
+				Lock:            lock(identity),
+				LeaseDuration:   5 * time.Second,
+				RenewDeadline:   3 * time.Second,
+				RetryPeriod:     1 * time.Second,
+				ReleaseOnCancel: true,
+				Callbacks: leaderelection.LeaderCallbacks{
+					OnStartedLeading: func(ctx context.Context) {
+						close(elected)
+						<-ctx.Done()
+					},
+					OnStoppedLeading: func() {},
+				},
+			})
+			if err != nil {
+				t.Errorf("NewLeaderElector(%s): %v", identity, err)
+				return
+			}
+
+			go le.Run(ctx)
+
+			select {
+			case <-elected:
+				results <- result{identity, true}
+				// Keep holding until context expires (8s total).
+				// Winner does NOT release early, so loser cannot
+				// re-acquire within its own 8s window.
+				<-ctx.Done()
+			case <-ctx.Done():
+				results <- result{identity, false}
+			}
+		}(id)
+	}
+
+	wg.Wait()
+	close(results)
+
+	winners := 0
+	for r := range results {
+		if r.elected {
+			winners++
+			t.Logf("winner: %s", r.identity)
+		}
+	}
+
+	if winners != 1 {
+		t.Fatalf("expected exactly 1 winner, got %d", winners)
+	}
+}
+
+// TestCancelElection tests that cancelling an election makes WaitLost return.
+func TestCancelElection(t *testing.T) {
+	ns := "default"
+	lease := "cancel-test"
+
+	err := runElection(ns, lease, "cancel-node:8080", 5, 4, 1)
+	if err != nil {
+		t.Fatalf("runElection failed: %v", err)
+	}
+
+	key := electionKey(ns, lease)
+	electionMutex.Lock()
+	state := elections[key]
+	electionMutex.Unlock()
+
+	// Wait for elected
+	select {
+	case <-state.elected:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out waiting for election")
+	}
+
+	// Cancel
+	state.cancel()
+
+	// WaitLost should return promptly
+	select {
+	case <-state.lost:
+		// success
+	case <-time.After(10 * time.Second):
+		t.Fatal("WaitLost did not return after cancel")
+	}
+}
+
+// TestGetHolderDuringElection verifies getHolder works while election is active.
+func TestGetHolderDuringElection(t *testing.T) {
+	ns := "default"
+	lease := "active-get-holder-test"
+	identity := "active-node:8080"
+
+	err := runElection(ns, lease, identity, 5, 4, 1)
+	if err != nil {
+		t.Fatalf("runElection failed: %v", err)
+	}
+
+	key := electionKey(ns, lease)
+	electionMutex.Lock()
+	state := elections[key]
+	electionMutex.Unlock()
+
+	select {
+	case <-state.elected:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out waiting for election")
+	}
+
+	// Concurrent getHolder calls during active election
+	var wg sync.WaitGroup
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			holder, _, err := getHolder(ns, lease)
+			if err != nil {
+				t.Errorf("getHolder during election failed: %v", err)
+				return
+			}
+			if holder != identity {
+				t.Errorf("expected %q, got %q", identity, holder)
+			}
+		}()
+	}
+	wg.Wait()
+
+	state.cancel()
+	<-state.lost
+}
+
+// TestGetHolderReturnsEmptyAfterLeaderDeath verifies that after a leader stops
+// renewing its lease without releasing it, getHolder returns an empty holder
+// once the lease expires. This is the integration-level counterpart to the
+// unit test TestGetHolderReturnsEmptyForExpiredLease.
+func TestGetHolderReturnsEmptyAfterLeaderDeath(t *testing.T) {
+	ns := "default"
+	lease := "expired-leader-test"
+	identity := "doomed-leader:8080"
+
+	// Acquire leadership without ReleaseOnCancel so canceling simulates a dead
+	// leader that stops renewing and leaves the old holder until expiry.
+	state, err := runElectionWithoutRelease(ns, lease, identity, 5, 4, 1)
+	if err != nil {
+		t.Fatalf("runElection failed: %v", err)
+	}
+
+	select {
+	case <-state.elected:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out waiting for election")
+	}
+
+	// Verify holder while active.
+	holder, _, err := getHolder(ns, lease)
+	if err != nil {
+		t.Fatalf("getHolder (active) failed: %v", err)
+	}
+	if holder != identity {
+		t.Fatalf("expected active holder %q, got %q", identity, holder)
+	}
+
+	// Simulate leader death: stop renewing without explicitly releasing.
+	state.cancel()
+	select {
+	case <-state.lost:
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for loss")
+	}
+
+	// Wait for the lease to expire (leaseDuration=5s, add margin).
+	time.Sleep(7 * time.Second)
+
+	// After expiry, getHolder must return empty holder so that the
+	// supervisor will attempt acquisition.
+	holder, _, err = getHolder(ns, lease)
+	if err != nil {
+		t.Fatalf("getHolder (expired) failed: %v", err)
+	}
+	if holder != "" {
+		t.Errorf("expected empty holder after lease expiry, got %q", holder)
+	}
+}
+
+// TestFailoverAfterLeaderDeath verifies that a new candidate can acquire
+// leadership after the previous leader dies and its lease expires.
+func TestFailoverAfterLeaderDeath(t *testing.T) {
+	ns := "default"
+	lease := "failover-test"
+
+	// First leader acquires without ReleaseOnCancel so canceling leaves the
+	// old holder in place until the lease naturally expires.
+	state1, err := runElectionWithoutRelease(ns, lease, "leader-1:8080", 5, 4, 1)
+	if err != nil {
+		t.Fatalf("first runElection failed: %v", err)
+	}
+
+	select {
+	case <-state1.elected:
+	case <-time.After(15 * time.Second):
+		t.Fatal("timed out waiting for first election")
+	}
+
+	// Simulate crash: cancel without release, wait for expiry.
+	state1.cancel()
+	<-state1.lost
+	time.Sleep(7 * time.Second)
+
+	// Second candidate should be able to acquire.
+	err = runElection(ns, lease, "leader-2:8080", 5, 4, 1)
+	if err != nil {
+		t.Fatalf("second runElection failed: %v", err)
+	}
+
+	key := electionKey(ns, lease)
+	electionMutex.Lock()
+	state2 := elections[key]
+	electionMutex.Unlock()
+
+	select {
+	case <-state2.elected:
+		// success — failover worked
+	case <-time.After(15 * time.Second):
+		t.Fatal("second candidate failed to acquire after leader death")
+	}
+
+	holder, _, err := getHolder(ns, lease)
+	if err != nil {
+		t.Fatalf("getHolder after failover failed: %v", err)
+	}
+	if holder != "leader-2:8080" {
+		t.Errorf("expected new leader %q, got %q", "leader-2:8080", holder)
+	}
+
+	state2.cancel()
+	<-state2.lost
+}

mooncake-common/k8s-lease/k8s_lease_wrapper.go

@@ -0,0 +1,489 @@
+package main
+
+/*
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+
+// Trampoline to invoke C/C++ callback safely from Go via cgo.
+typedef void (*holder_change_cb_t)(void* ctx,
+                                    const char* holder, size_t holderSize,
+                                    int64_t leaseTransitions);
+
+static inline void call_holder_change_cb(holder_change_cb_t func, void* ctx,
+                                          const char* holder, size_t holderSize,
+                                          int64_t leaseTransitions) {
+  func(ctx, holder, holderSize, leaseTransitions);
+}
+*/
+import "C"
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"time"
+	"unsafe"
+
+	coordinationv1 "k8s.io/api/coordination/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+)
+
+// electionState holds the runtime state for a single leader election.
+type electionState struct {
+	cancel      context.CancelFunc
+	elected     chan struct{} // closed when OnStartedLeading fires
+	lost        chan struct{} // closed when OnStoppedLeading fires
+	err         error         // set before lost is closed, if any
+	transitions int64         // set before elected is closed
+}
+
+// watchState holds the runtime state for a single Lease watch.
+type watchState struct {
+	cancel context.CancelFunc
+}
+
+var (
+	globalClient kubernetes.Interface
+	clientMutex  sync.Mutex
+	initClientFn = initClient
+
+	elections     = make(map[string]*electionState)
+	electionMutex sync.Mutex
+
+	watches    = make(map[string]*watchState)
+	watchMutex sync.Mutex
+)
+
+func electionKey(namespace, leaseName string) string {
+	return namespace + "/" + leaseName
+}
+
+func ensureClientInitialized() error {
+	clientMutex.Lock()
+	initialized := globalClient != nil
+	clientMutex.Unlock()
+	if initialized {
+		return nil
+	}
+	return initClientFn()
+}
+
+// initClient creates the K8s clientset from in-cluster config or KUBECONFIG.
+func initClient() error {
+	clientMutex.Lock()
+	defer clientMutex.Unlock()
+	if globalClient != nil {
+		return nil
+	}
+
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		// Fall back to KUBECONFIG
+		kubeconfig := os.Getenv("KUBECONFIG")
+		if kubeconfig == "" {
+			home := os.Getenv("HOME")
+			if home != "" {
+				kubeconfig = home + "/.kube/config"
+			}
+		}
+		config, err = clientcmd.BuildConfigFromFlags("", kubeconfig)
+		if err != nil {
+			return fmt.Errorf("failed to build k8s config: %w", err)
+		}
+	}
+
+	client, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return fmt.Errorf("failed to create k8s clientset: %w", err)
+	}
+	globalClient = client
+	return nil
+}
+
+// runElection starts a leader election goroutine for the given namespace/leaseName.
+func runElection(namespace, leaseName, identity string,
+	leaseDurationSec, renewDeadlineSec, retryPeriodSec int) error {
+	if err := ensureClientInitialized(); err != nil {
+		return err
+	}
+
+	key := electionKey(namespace, leaseName)
+
+	electionMutex.Lock()
+	if _, exists := elections[key]; exists {
+		electionMutex.Unlock()
+		return fmt.Errorf("election already running for %s", key)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	state := &electionState{
+		cancel:  cancel,
+		elected: make(chan struct{}),
+		lost:    make(chan struct{}),
+	}
+	elections[key] = state
+	electionMutex.Unlock()
+
+	lock := &resourcelock.LeaseLock{
+		LeaseMeta: metav1.ObjectMeta{
+			Name:      leaseName,
+			Namespace: namespace,
+		},
+		Client: globalClient.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: identity,
+		},
+	}
+
+	le, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
+		Lock:            lock,
+		LeaseDuration:   time.Duration(leaseDurationSec) * time.Second,
+		RenewDeadline:   time.Duration(renewDeadlineSec) * time.Second,
+		RetryPeriod:     time.Duration(retryPeriodSec) * time.Second,
+		ReleaseOnCancel: true,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(ctx context.Context) {
+				_, transitions, err := getHolder(namespace, leaseName)
+				if err == nil {
+					state.transitions = transitions
+				}
+				close(state.elected)
+				// Block until context is cancelled (leadership lost or explicit cancel)
+				<-ctx.Done()
+			},
+			OnStoppedLeading: func() {
+				close(state.lost)
+				// Auto-cleanup: remove from map so the same key can be reused.
+				electionMutex.Lock()
+				if elections[key] == state {
+					delete(elections, key)
+				}
+				electionMutex.Unlock()
+			},
+		},
+	})
+	if err != nil {
+		electionMutex.Lock()
+		delete(elections, key)
+		electionMutex.Unlock()
+		cancel()
+		return fmt.Errorf("failed to create leader elector: %w", err)
+	}
+
+	go le.Run(ctx)
+	return nil
+}
+
+// getHolder reads the current Lease holder identity and transitions.
+func getHolder(namespace, leaseName string) (string, int64, error) {
+	if err := ensureClientInitialized(); err != nil {
+		return "", 0, err
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	lease, err := globalClient.CoordinationV1().Leases(namespace).Get(ctx, leaseName, metav1.GetOptions{})
+	if err != nil {
+		return "", 0, fmt.Errorf("failed to get lease: %w", err)
+	}
+
+	holder := ""
+	if lease.Spec.HolderIdentity != nil {
+		holder = *lease.Spec.HolderIdentity
+	}
+	transitions := int64(0)
+	if lease.Spec.LeaseTransitions != nil {
+		transitions = int64(*lease.Spec.LeaseTransitions)
+	}
+
+	// Treat expired leases as having no holder so that the C++ supervisor
+	// will attempt acquisition instead of going to standby.
+	if holder != "" && lease.Spec.RenewTime != nil && lease.Spec.LeaseDurationSeconds != nil {
+		expiry := lease.Spec.RenewTime.Time.Add(time.Duration(*lease.Spec.LeaseDurationSeconds) * time.Second)
+		if time.Now().After(expiry) {
+			holder = ""
+		}
+	}
+	return holder, transitions, nil
+}
+
+//export K8sLeaseInit
+func K8sLeaseInit(errMsg **C.char) C.int {
+	if err := ensureClientInitialized(); err != nil {
+		*errMsg = C.CString(err.Error())
+		return -1
+	}
+	return 0
+}
+
+//export K8sLeaseRunElection
+func K8sLeaseRunElection(
+	ns, leaseName, identity *C.char,
+	leaseDurationSec, renewDeadlineSec, retryPeriodSec C.int,
+	errMsg **C.char,
+) C.int {
+	nsStr := C.GoString(ns)
+	ln := C.GoString(leaseName)
+	id := C.GoString(identity)
+
+	err := runElection(nsStr, ln, id,
+		int(leaseDurationSec), int(renewDeadlineSec), int(retryPeriodSec))
+	if err != nil {
+		*errMsg = C.CString(err.Error())
+		return -1
+	}
+	return 0
+}
+
+//export K8sLeaseWaitElected
+func K8sLeaseWaitElected(
+	ns, leaseName *C.char,
+	timeoutSec C.int,
+	leaseTransitions *C.longlong,
+	errMsg **C.char,
+) C.int {
+	key := electionKey(C.GoString(ns), C.GoString(leaseName))
+
+	electionMutex.Lock()
+	state, exists := elections[key]
+	electionMutex.Unlock()
+
+	if !exists {
+		*errMsg = C.CString("no election running for " + key)
+		return -1
+	}
+
+	timeout := time.Duration(timeoutSec) * time.Second
+
+	// Wait for elected, lost, or timeout
+	select {
+	case <-state.elected:
+		*leaseTransitions = C.longlong(state.transitions)
+		return 0
+	case <-state.lost:
+		*errMsg = C.CString("election lost before becoming leader")
+		return -1
+	case <-time.After(timeout):
+		state.cancel()
+		<-state.lost
+		*errMsg = C.CString("election timed out after " + fmt.Sprintf("%d", int(timeoutSec)) + "s")
+		return -1
+	}
+}
+
+//export K8sLeaseWaitLost
+func K8sLeaseWaitLost(
+	ns, leaseName *C.char,
+	errMsg **C.char,
+) C.int {
+	key := electionKey(C.GoString(ns), C.GoString(leaseName))
+
+	electionMutex.Lock()
+	state, exists := elections[key]
+	electionMutex.Unlock()
+
+	if !exists {
+		// Already cleaned up by OnStoppedLeading — election is over.
+		return 0
+	}
+
+	<-state.lost
+
+	if state.err != nil {
+		*errMsg = C.CString(state.err.Error())
+		return -1
+	}
+	return 0
+}
+
+//export K8sLeaseCancelElection
+func K8sLeaseCancelElection(
+	ns, leaseName *C.char,
+	errMsg **C.char,
+) C.int {
+	key := electionKey(C.GoString(ns), C.GoString(leaseName))
+
+	electionMutex.Lock()
+	state, exists := elections[key]
+	electionMutex.Unlock()
+
+	if !exists {
+		// Idempotent — no error if no election
+		return 0
+	}
+
+	state.cancel()
+	return 0
+}
+
+//export K8sLeaseGetHolder
+func K8sLeaseGetHolder(
+	ns, leaseName *C.char,
+	holderIdentity **C.char,
+	leaseTransitions *C.longlong,
+	errMsg **C.char,
+) C.int {
+	nsStr := C.GoString(ns)
+	ln := C.GoString(leaseName)
+
+	holder, transitions, err := getHolder(nsStr, ln)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			*holderIdentity = nil
+			*leaseTransitions = 0
+			return 1
+		}
+		errStr := err.Error()
+		*errMsg = C.CString(errStr)
+		return -1
+	}
+
+	if holder == "" {
+		*holderIdentity = nil
+	} else {
+		*holderIdentity = C.CString(holder)
+	}
+	*leaseTransitions = C.longlong(transitions)
+	return 0
+}
+
+//export K8sLeaseWatchHolder
+func K8sLeaseWatchHolder(
+	ns, leaseName *C.char,
+	callbackCtx unsafe.Pointer,
+	callbackFunc C.holder_change_cb_t,
+	errMsg **C.char,
+) C.int {
+	nsStr := C.GoString(ns)
+	ln := C.GoString(leaseName)
+	key := electionKey(nsStr, ln)
+
+	if callbackFunc == nil {
+		*errMsg = C.CString("callback function is nil")
+		return -1
+	}
+	if err := ensureClientInitialized(); err != nil {
+		*errMsg = C.CString(err.Error())
+		return -1
+	}
+
+	watchMutex.Lock()
+	if _, exists := watches[key]; exists {
+		watchMutex.Unlock()
+		*errMsg = C.CString("watch already running for " + key)
+		return -1
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	watches[key] = &watchState{cancel: cancel}
+	watchMutex.Unlock()
+
+	go func() {
+		defer func() {
+			watchMutex.Lock()
+			delete(watches, key)
+			watchMutex.Unlock()
+		}()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+			}
+
+			watcher, err := globalClient.CoordinationV1().Leases(nsStr).Watch(ctx, metav1.ListOptions{
+				FieldSelector: "metadata.name=" + ln,
+			})
+			if err != nil {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+					time.Sleep(time.Second)
+					continue
+				}
+			}
+
+			for event := range watcher.ResultChan() {
+				select {
+				case <-ctx.Done():
+					watcher.Stop()
+					return
+				default:
+				}
+
+				if event.Type == watch.Modified || event.Type == watch.Added {
+					lease, ok := event.Object.(*coordinationv1.Lease)
+					if !ok {
+						continue
+					}
+					holder := ""
+					if lease.Spec.HolderIdentity != nil {
+						holder = *lease.Spec.HolderIdentity
+					}
+					transitions := int64(0)
+					if lease.Spec.LeaseTransitions != nil {
+						transitions = int64(*lease.Spec.LeaseTransitions)
+					}
+
+					var holderPtr *C.char
+					var holderSize C.size_t
+					if holder != "" {
+						holderPtr = C.CString(holder)
+						holderSize = C.size_t(len(holder))
+					}
+
+					C.call_holder_change_cb(callbackFunc, callbackCtx,
+						holderPtr, holderSize, C.int64_t(transitions))
+
+					if holderPtr != nil {
+						C.free(unsafe.Pointer(holderPtr))
+					}
+				}
+			}
+
+			// Watch channel closed — retry unless cancelled
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				time.Sleep(time.Second)
+			}
+		}
+	}()
+
+	return 0
+}
+
+//export K8sLeaseCancelWatch
+func K8sLeaseCancelWatch(
+	ns, leaseName *C.char,
+	errMsg **C.char,
+) C.int {
+	key := electionKey(C.GoString(ns), C.GoString(leaseName))
+
+	watchMutex.Lock()
+	state, exists := watches[key]
+	watchMutex.Unlock()
+
+	if !exists {
+		// Idempotent
+		return 0
+	}
+
+	state.cancel()
+	return 0
+}
+
+func main() {}

mooncake-common/k8s-lease/k8s_lease_wrapper_test.go

@@ -0,0 +1,293 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	coordinationv1 "k8s.io/api/coordination/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/utils/ptr"
+)
+
+// swapClient replaces globalClient and returns the old one.
+func swapClient(newClient kubernetes.Interface) kubernetes.Interface {
+	clientMutex.Lock()
+	defer clientMutex.Unlock()
+	old := globalClient
+	globalClient = newClient
+	return old
+}
+
+// TestGetHolderWithFakeClient tests getHolder using a fake K8s clientset.
+func TestGetHolderWithFakeClient(t *testing.T) {
+	holderID := "node-1:8080"
+	transitions := int32(3)
+	lease := &coordinationv1.Lease{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-lease",
+			Namespace: "default",
+		},
+		Spec: coordinationv1.LeaseSpec{
+			HolderIdentity:   &holderID,
+			LeaseTransitions: &transitions,
+		},
+	}
+
+	fakeClient := fake.NewSimpleClientset(lease)
+	old := swapClient(fakeClient)
+	defer swapClient(old)
+
+	holder, trans, err := getHolder("default", "test-lease")
+	if err != nil {
+		t.Fatalf("getHolder failed: %v", err)
+	}
+	if holder != holderID {
+		t.Errorf("expected holder %q, got %q", holderID, holder)
+	}
+	if trans != int64(transitions) {
+		t.Errorf("expected transitions %d, got %d", transitions, trans)
+	}
+}
+
+// TestGetHolderNotFound tests getHolder when the Lease does not exist.
+func TestGetHolderNotFound(t *testing.T) {
+	fakeClient := fake.NewSimpleClientset()
+	old := swapClient(fakeClient)
+	defer swapClient(old)
+
+	_, _, err := getHolder("default", "nonexistent")
+	if err == nil {
+		t.Fatal("expected error for nonexistent lease, got nil")
+	}
+}
+
+// TestGetHolderEmptyIdentity tests getHolder when holder is nil.
+func TestGetHolderEmptyIdentity(t *testing.T) {
+	lease := &coordinationv1.Lease{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "empty-lease",
+			Namespace: "default",
+		},
+		Spec: coordinationv1.LeaseSpec{},
+	}
+	fakeClient := fake.NewSimpleClientset(lease)
+	old := swapClient(fakeClient)
+	defer swapClient(old)
+
+	holder, trans, err := getHolder("default", "empty-lease")
+	if err != nil {
+		t.Fatalf("getHolder failed: %v", err)
+	}
+	if holder != "" {
+		t.Errorf("expected empty holder, got %q", holder)
+	}
+	if trans != 0 {
+		t.Errorf("expected 0 transitions, got %d", trans)
+	}
+}
+
+// TestGetHolderReturnsEmptyForExpiredLease verifies that getHolder treats a
+// lease whose renewTime + leaseDuration is in the past as having no holder.
+// This is critical for failover: when a leader pod dies without releasing the
+// lease, standbys must see an empty holder so the supervisor attempts
+// acquisition instead of looping in standby.
+func TestGetHolderReturnsEmptyForExpiredLease(t *testing.T) {
+	holderID := "dead-leader:8080"
+	leaseDuration := int32(5)
+	transitions := int32(2)
+	expiredRenewTime := metav1.NewMicroTime(time.Now().Add(-10 * time.Second))
+
+	lease := &coordinationv1.Lease{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "expired-lease",
+			Namespace: "default",
+		},
+		Spec: coordinationv1.LeaseSpec{
+			HolderIdentity:       &holderID,
+			LeaseDurationSeconds: &leaseDuration,
+			LeaseTransitions:     &transitions,
+			RenewTime:            &expiredRenewTime,
+		},
+	}
+
+	fakeClient := fake.NewSimpleClientset(lease)
+	old := swapClient(fakeClient)
+	defer swapClient(old)
+
+	holder, trans, err := getHolder("default", "expired-lease")
+	if err != nil {
+		t.Fatalf("getHolder failed: %v", err)
+	}
+	if holder != "" {
+		t.Errorf("expected empty holder for expired lease, got %q", holder)
+	}
+	// Transitions should still be reported even for expired leases.
+	if trans != int64(transitions) {
+		t.Errorf("expected transitions %d, got %d", transitions, trans)
+	}
+}
+
+// TestGetHolderReturnsHolderForActiveLease verifies that getHolder returns the
+// holder identity when the lease is still active (renewTime + leaseDuration is
+// in the future).
+func TestGetHolderReturnsHolderForActiveLease(t *testing.T) {
+	holderID := "active-leader:8080"
+	leaseDuration := int32(15)
+	transitions := int32(1)
+	recentRenewTime := metav1.NewMicroTime(time.Now())
+
+	lease := &coordinationv1.Lease{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "active-lease",
+			Namespace: "default",
+		},
+		Spec: coordinationv1.LeaseSpec{
+			HolderIdentity:       &holderID,
+			LeaseDurationSeconds: &leaseDuration,
+			LeaseTransitions:     &transitions,
+			RenewTime:            &recentRenewTime,
+		},
+	}
+
+	fakeClient := fake.NewSimpleClientset(lease)
+	old := swapClient(fakeClient)
+	defer swapClient(old)
+
+	holder, trans, err := getHolder("default", "active-lease")
+	if err != nil {
+		t.Fatalf("getHolder failed: %v", err)
+	}
+	if holder != holderID {
+		t.Errorf("expected holder %q, got %q", holderID, holder)
+	}
+	if trans != int64(transitions) {
+		t.Errorf("expected transitions %d, got %d", transitions, trans)
+	}
+}
+
+// TestElectionKeyFormat tests the election key construction.
+func TestElectionKeyFormat(t *testing.T) {
+	tests := []struct {
+		ns, name, want string
+	}{
+		{"default", "leader", "default/leader"},
+		{"kube-system", "my-lock", "kube-system/my-lock"},
+		{"", "bare", "/bare"},
+	}
+	for _, tc := range tests {
+		got := electionKey(tc.ns, tc.name)
+		if got != tc.want {
+			t.Errorf("electionKey(%q, %q) = %q, want %q", tc.ns, tc.name, got, tc.want)
+		}
+	}
+}
+
+// TestLeaseCRUDWithFakeClient tests basic Lease CRUD via the K8s API.
+func TestLeaseCRUDWithFakeClient(t *testing.T) {
+	fakeClient := fake.NewSimpleClientset()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	holderID := "node-a:9090"
+	transitions := int32(0)
+	lease := &coordinationv1.Lease{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "crud-test",
+			Namespace: "default",
+		},
+		Spec: coordinationv1.LeaseSpec{
+			HolderIdentity:   &holderID,
+			LeaseTransitions: &transitions,
+		},
+	}
+	created, err := fakeClient.CoordinationV1().Leases("default").Create(ctx, lease, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatalf("create lease failed: %v", err)
+	}
+	if *created.Spec.HolderIdentity != holderID {
+		t.Errorf("created holder = %q, want %q", *created.Spec.HolderIdentity, holderID)
+	}
+
+	newHolder := "node-b:9090"
+	newTransitions := int32(1)
+	created.Spec.HolderIdentity = &newHolder
+	created.Spec.LeaseTransitions = &newTransitions
+	updated, err := fakeClient.CoordinationV1().Leases("default").Update(ctx, created, metav1.UpdateOptions{})
+	if err != nil {
+		t.Fatalf("update lease failed: %v", err)
+	}
+	if *updated.Spec.HolderIdentity != newHolder {
+		t.Errorf("updated holder = %q, want %q", *updated.Spec.HolderIdentity, newHolder)
+	}
+	if *updated.Spec.LeaseTransitions != newTransitions {
+		t.Errorf("updated transitions = %d, want %d", *updated.Spec.LeaseTransitions, newTransitions)
+	}
+
+	got, err := fakeClient.CoordinationV1().Leases("default").Get(ctx, "crud-test", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("get lease failed: %v", err)
+	}
+	if *got.Spec.HolderIdentity != newHolder {
+		t.Errorf("got holder = %q, want %q", *got.Spec.HolderIdentity, newHolder)
+	}
+
+	err = fakeClient.CoordinationV1().Leases("default").Delete(ctx, "crud-test", metav1.DeleteOptions{})
+	if err != nil {
+		t.Fatalf("delete lease failed: %v", err)
+	}
+
+	_, err = fakeClient.CoordinationV1().Leases("default").Get(ctx, "crud-test", metav1.GetOptions{})
+	if err == nil {
+		t.Fatal("expected error after delete, got nil")
+	}
+}
+
+// TestConcurrentGetHolder tests concurrent calls to getHolder.
+func TestConcurrentGetHolder(t *testing.T) {
+	holderID := "concurrent-node:8080"
+	lease := &coordinationv1.Lease{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "concurrent-lease",
+			Namespace: "default",
+		},
+		Spec: coordinationv1.LeaseSpec{
+			HolderIdentity:   &holderID,
+			LeaseTransitions: ptr.To(int32(5)),
+		},
+	}
+	fakeClient := fake.NewSimpleClientset(lease)
+	old := swapClient(fakeClient)
+	defer swapClient(old)
+
+	const n = 10
+	errCh := make(chan error, n)
+	for i := 0; i < n; i++ {
+		go func() {
+			holder, trans, err := getHolder("default", "concurrent-lease")
+			if err != nil {
+				errCh <- err
+				return
+			}
+			if holder != holderID {
+				errCh <- fmt.Errorf("expected holder %q, got %q", holderID, holder)
+				return
+			}
+			if trans != 5 {
+				errCh <- fmt.Errorf("expected transitions 5, got %d", trans)
+				return
+			}
+			errCh <- nil
+		}()
+	}
+
+	for i := 0; i < n; i++ {
+		if err := <-errCh; err != nil {
+			t.Fatalf("concurrent getHolder failed: %v", err)
+		}
+	}
+}

mooncake-common/src/CMakeLists.txt

@@ -1,13 +1,64 @@
 find_package(yaml-cpp REQUIRED)
 
+find_package(asio QUIET)
+
+if(asio_FOUND)
+    message(STATUS "Found ASIO via find_package")
+    set(ASIO_INCLUDE_DIR ${asio_INCLUDE_DIR})
+else()
+    find_path(ASIO_INCLUDE_DIR
+        NAMES asio.hpp
+        PATHS
+            /usr/local/include
+            /usr/include
+            ${CMAKE_INSTALL_PREFIX}/include
+        DOC "Path to ASIO headers"
+    )
+
+    if(NOT ASIO_INCLUDE_DIR)
+        message(FATAL_ERROR "ASIO not found. Please install ASIO or set ASIO_INCLUDE_DIR manually.")
+    endif()
+
+    message(STATUS "Found ASIO at: ${ASIO_INCLUDE_DIR}")
+endif()
+
 set(MOONCAKE_COMMON_SOURCES
     default_config.cpp
     environ.cpp
 )
+
+add_library(asio_shared SHARED asio_impl.cpp)
+
+target_compile_definitions(asio_shared
+    PUBLIC
+        ASIO_SEPARATE_COMPILATION
+        ASIO_DYN_LINK
+)
+
+target_include_directories(asio_shared
+    PUBLIC
+        ${ASIO_INCLUDE_DIR}
+)
+
+set_target_properties(asio_shared PROPERTIES
+    POSITION_INDEPENDENT_CODE ON
+    INSTALL_RPATH "$ORIGIN"
+    BUILD_WITH_INSTALL_RPATH TRUE
+    OUTPUT_NAME "asio"
+    LIBRARY_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/mooncake-common"
+)
+
+target_link_libraries(asio_shared PUBLIC pthread)
+
 add_library(mooncake_common
     ${MOONCAKE_COMMON_SOURCES}
 )
 
+target_include_directories(mooncake_common PUBLIC
+    $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/../include>
+    $<INSTALL_INTERFACE:include>
+)
+
 target_link_libraries(mooncake_common PUBLIC
     yaml-cpp
     jsoncpp
@@ -16,3 +67,5 @@ target_link_libraries(mooncake_common PUBLIC
 if (BUILD_SHARED_LIBS)
   install(TARGETS mooncake_common DESTINATION lib)
 endif()
+
+install(TARGETS asio_shared DESTINATION lib)

mooncake-p2p-store/build.sh

@@ -33,7 +33,7 @@ fi
 
 EXT_LDFLAGS="-L$BUILD_DIR/mooncake-transfer-engine/src"
 EXT_LDFLAGS+=" -L$BUILD_DIR/mooncake-transfer-engine/src/common/base"
-EXT_LDFLAGS+=" -L$BUILD_DIR/mooncake-asio"
+EXT_LDFLAGS+=" -L$BUILD_DIR/mooncake-common"
 EXT_LDFLAGS+=" -ltransfer_engine -lbase -lasio -lstdc++ -lnuma -lglog -libverbs -ljsoncpp"
 
 if [ -d "/usr/local/cuda/lib64/stubs" ]; then

mooncake-pg/include/mooncake_worker.cuh

@@ -53,7 +53,7 @@ __global__ struct Task {
     size_t tensorSize;  // In bytes
     int64_t broadcastRoot;
     int bufferOffset;
-    uint64_t submit_sequence = 0;
+    uint64_t submitSequence = 0;
     BatchID batchID;
     void* transferGroupMeta;
 };

mooncake-pg/src/mooncake_worker.cu

@@ -3,6 +3,7 @@
 #include <memory>
 #include <thread>
 #include <mooncake_worker.cuh>
+#include <ATen/cuda/CUDAGraphsUtils.cuh>
 
 #include "pg_utils.h"
 
@@ -45,36 +46,68 @@ class MooncakeWorkCuda : public ::c10d::Work {
 
     bool wait(std::chrono::milliseconds timeout) override {
         // Wait until the task has been submitted to TransferEngine:
-        // This ensures that the CUDA kernels required for the transfer
+        // This tries to ensure that the CUDA kernels required for the transfer
         // have been launched by the time `waitUntilTasksSubmitted` returns.
-        // Although this is not required for CPU-only transports such as
-        // RdmaTransport, we keep this behavior to avoid invasive changes
-        // to TE/TENT.
+        //
+        // Why is this needed? PyTorch documentation implies that collective
+        // operations should be enqueued when `wait()` returns. In practice, we
+        // found that violating this causes hangs.
+        //
+        // Our current hypothesis for the hang is: PyTorch assumes the kernels
+        // needed for the transfer are already launched when `wait` returns
+        // true. It may then launch subsequent operations after the collective
+        // (e.g., `.cpu()`). Such operations may acquire a process-wide lock in
+        // the CUDA runtime. Also, they may rely on the data produced by the
+        // collective, thus causing a synchronization on enq_stream. However,
+        // holding that runtime lock prevents cudaMemcpy(Async) in TE/TENT from
+        // launching. This means the transfer can't finish, and enq_stream won't
+        // complete. Thus, a deadlock occurs.
+        // (In practice, we found that replacing all cudaMemcpyAsync in TENT
+        // with cuMemcpyAsync actually alleviates this, which further suggests a
+        // deadlock in the CUDA runtime. However, that change is too invasive
+        // for TE/TENT, so we do not adopt it here.)
+        //
+        // Strictly speaking, the wait is needed for another reason: The current
+        // stream will be blocked on the event below. Any subsequent work on
+        // `current_stream` will wait on that event, which effectively waits for
+        // the task to be done. Therefore, we must ensure all kernels needed for
+        // the transfer task are launched BEFORE blocking the current stream, in
+        // case TE/TENT use `current_stream` to launch those kernels (though it
+        // is rare).
         //
         // Please note that this logic relies on the assumption that TE/TENT
         // will launch all CUDA operations in `submitTransfer`.
         // Unfortunately, TcpTransport in TE and TENT currently violates this
-        // assumption (cudaMemcpy(Async) may be called from a callback), which
-        // can cause hangs in PG when a CUDA operation such as `x.cpu().item()`
-        // follows the collective. For TE's TcpTransport, the use of cudaMemcpy
-        // on the default stream may also contribute to the hang.
+        // assumption (cudaMemcpy(Async) may be called later from a callback),
+        // which can cause hangs in PG when a CUDA operation such as
+        // `x.cpu().item()` follows the collective. For TE's TcpTransport, the
+        // use of cudaMemcpy on the default stream may also contribute to the
+        // hang.
         //
-        // This wait is primarily needed for two reasons:
-        //   1. PyTorch documents that `wait` should ensure the operation is
-        //      issued, though not necessarily completed, for CUDA work.
-        //      In practice, this means the transfer kernel must at least be
-        //      launched; otherwise, a hang may occur.
-        //   2. The current stream is blocked on the event below. Any subsequent
-        //      work on `current_stream` will wait on that event, which
-        //      effectively waits for the task to be done. Therefore, we must
-        //      not block until all kernels needed for the transfer task have
-        //      been launched, in case TE/TENT use `current_stream` to
-        //      launch those kernels (though rare).
-        auto submitted =
-            worker_->waitUntilTasksSubmitted(submitted_tasks_, timeout);
+        // Besides, for CPU-only transports (like RdmaTransport),
+        // waitUntilTasksSubmitted is totally unnecessary, but we keep it for
+        // uniform behavior to avoid invasive changes to TE/TENT.
+        bool submitted = true;
+        if (at::cuda::currentStreamCaptureStatus() ==
+            c10::cuda::CaptureStatus::None) {
+            // Normal execution: block until tasks are submitted.
+            submitted =
+                worker_->waitUntilTasksSubmitted(submitted_tasks_, timeout);
+        } else {
+            // During CUDA graph capture, kernels are recorded but not actually
+            // executed. The enqueueTaskKernel would never run, so
+            // waitUntilTasksSubmitted would hang because the CPU worker thread
+            // never sees task.active == true.
+            //
+            // Note that this also means NvlinkTransport (and TcpTransport too,
+            // of course) won't work with CUDA Graphs: Kernels launched inside
+            // TE/TENT can't be captured by the graph, and during replay they
+            // are not ordered with the graph execution. This may trigger the
+            // same deadlock described above.
+        }
         if (!submitted) return false;
 
-        // Once all tasks have been submitted, create an event to synchronize
+        // Once all tasks have been submitted, use the event to synchronize
         // the current stream and the enqueue stream, but do not wait on this
         // event.
         //
@@ -101,6 +134,18 @@ class MooncakeBarrierWorkCuda : public MooncakeWorkCuda {
     using MooncakeWorkCuda::MooncakeWorkCuda;
 
     bool wait(std::chrono::milliseconds timeout) override {
+        // Skip host-side synchronization during CUDA graph capture.
+        // cudaEventSynchronize is not permitted while a stream is capturing.
+        if (at::cuda::currentStreamCaptureStatus() !=
+            c10::cuda::CaptureStatus::None) {
+            // We still need stream-level synchronization so that subsequent
+            // operations on the capture stream are ordered after the barrier
+            // task on the enqueue stream.
+            auto current_stream = at::cuda::getCurrentCUDAStream();
+            event_->block(current_stream);
+            return true;
+        }
+
         if (timeout == kNoTimeout) {
             event_->synchronize();
             return true;
@@ -123,7 +168,7 @@ __global__ void enqueueTaskKernel(c10d::OpType opType, size_t tensorSize,
     tasks[taskId].tensorSize = tensorSize;
     tasks[taskId].broadcastRoot = broadcastRoot;
     tasks[taskId].bufferOffset = bufferOffset;
-    tasks[taskId].submit_sequence = submitSequence;
+    tasks[taskId].submitSequence = submitSequence;
     tasks[taskId].transferGroupMeta = meta;
 
     // Publish task metadata before notifying the host worker thread.
@@ -356,7 +401,7 @@ MooncakeWorker::MooncakeWorker(int cuda_device_index)
     }
     for (size_t i = 0; i < kNumTasks_; ++i) {
         tasks_[i].active = false;
-        tasks_[i].submit_sequence = 0;
+        tasks_[i].submitSequence = 0;
         submitted_task_sequence_[i].store(0, std::memory_order_relaxed);
     }
 }

mooncake-pg/src/mooncake_worker_thread.cpp

@@ -91,7 +91,7 @@ void MooncakeWorker::startWorker() {
                                      group->rank != task.broadcastRoot) ||
                                     task.opType == c10d::OpType::BARRIER;
                 if (task_status[i].load(std::memory_order_acquire) == IDLE) {
-                    const auto submit_sequence = task.submit_sequence;
+                    const auto submit_sequence = task.submitSequence;
                     if (skipTransfer) {
                         submitted_task_sequence_[i].store(
                             submit_sequence, std::memory_order_release);

mooncake-store/go/build.sh

@@ -39,7 +39,7 @@ CGO_LDFLAGS="-L${BUILD_DIR}/mooncake-store/src"
 CGO_LDFLAGS+=" -L${BUILD_DIR}/mooncake-store/src/cachelib_memory_allocator"
 CGO_LDFLAGS+=" -L${BUILD_DIR}/mooncake-transfer-engine/src"
 CGO_LDFLAGS+=" -L${BUILD_DIR}/mooncake-transfer-engine/src/common/base"
-CGO_LDFLAGS+=" -L${BUILD_DIR}/mooncake-asio"
+CGO_LDFLAGS+=" -L${BUILD_DIR}/mooncake-common"
 CGO_LDFLAGS+=" -lmooncake_store -lcachelib_memory_allocator -ltransfer_engine -lbase -lasio"
 CGO_LDFLAGS+=" -lstdc++ -lnuma -lglog -lgflags -libverbs -ljsoncpp -lzstd -lcurl"
 

mooncake-store/include/client_metric.h

@@ -1,10 +1,15 @@
 #pragma once
 
+#include <algorithm>
+#include <array>
 #include <atomic>
 #include <chrono>
 #include <iomanip>
+#include <mutex>
+#include <optional>
 #include <sstream>
 #include <thread>
+#include <unordered_set>
 #include <vector>
 #include <ylt/metric/counter.hpp>
 #include <ylt/metric/histogram.hpp>
@@ -46,6 +51,56 @@ const inline std::map<std::string, std::string> merge_labels(
     return merged_labels;
 }
 
+inline std::string format_metric_rate(double value, const char* suffix) {
+    const double KB = 1024.0;
+    const double MB = KB * 1024.0;
+    const double GB = MB * 1024.0;
+    const double TB = GB * 1024.0;
+
+    std::ostringstream oss;
+    oss << std::fixed << std::setprecision(2);
+    if (value >= TB) {
+        oss << value / TB << " T" << suffix;
+    } else if (value >= GB) {
+        oss << value / GB << " G" << suffix;
+    } else if (value >= MB) {
+        oss << value / MB << " M" << suffix;
+    } else if (value >= KB) {
+        oss << value / KB << " K" << suffix;
+    } else {
+        oss << value << " " << suffix;
+    }
+    return oss.str();
+}
+
+inline std::string format_metric_bandwidth(uint64_t total_bytes,
+                                           double elapsed_seconds) {
+    return format_metric_rate(total_bytes / elapsed_seconds, "B/s");
+}
+
+inline uint64_t elapsed_us_since(
+    std::chrono::steady_clock::time_point start_time) {
+    return static_cast<uint64_t>(
+        std::chrono::duration_cast<std::chrono::microseconds>(
+            std::chrono::steady_clock::now() - start_time)
+            .count());
+}
+
+template <typename Result, typename Operation, typename SuccessFn,
+          typename ObserveFn>
+Result execute_timed_operation(Operation&& operation, SuccessFn&& success_fn,
+                               ObserveFn&& observe_fn) {
+    const auto start_time = std::chrono::steady_clock::now();
+    Result result = std::forward<Operation>(operation)();
+    if (std::forward<SuccessFn>(success_fn)(result)) {
+        std::forward<ObserveFn>(observe_fn)(elapsed_us_since(start_time),
+                                            result);
+    }
+    return result;
+}
+
+enum class TransferOperationKind { kRead, kWrite };
+
 struct TransferMetric {
     TransferMetric(std::map<std::string, std::string> labels = {})
         : total_read_bytes("mooncake_transfer_read_bytes", "Total bytes read",
@@ -61,7 +116,8 @@ struct TransferMetric {
           get_latency_us("mooncake_transfer_get_latency",
                          "Get transfer latency (us)", kLatencyBucket, labels),
           put_latency_us("mooncake_transfer_put_latency",
-                         "Put transfer latency (us)", kLatencyBucket, labels) {}
+                         "Put transfer latency (us)", kLatencyBucket, labels),
+          start_time_(std::chrono::steady_clock::now()) {}
 
     ylt::metric::counter_t total_read_bytes;
     ylt::metric::counter_t total_write_bytes;
@@ -79,7 +135,7 @@ struct TransferMetric {
         put_latency_us.serialize(str);
     }
 
-    std::string summary_metrics() {
+    std::string summary_metrics(bool include_bandwidth = true) {
         std::stringstream ss;
         ss << "=== Transfer Metrics Summary ===\n";
 
@@ -88,6 +144,14 @@ struct TransferMetric {
         auto write_bytes = total_write_bytes.value();
         ss << "Total Read: " << byte_size_to_string(read_bytes) << "\n";
         ss << "Total Write: " << byte_size_to_string(write_bytes) << "\n";
+        if (include_bandwidth) {
+            ss << "Average Read Throughput: "
+               << format_metric_bandwidth(read_bytes, elapsed_seconds())
+               << "\n";
+            ss << "Average Write Throughput: "
+               << format_metric_bandwidth(write_bytes, elapsed_seconds())
+               << "\n";
+        }
 
         // Latency summaries
         ss << "\n=== Latency Summary (microseconds) ===\n";
@@ -102,6 +166,14 @@ struct TransferMetric {
     }
 
    private:
+    std::chrono::steady_clock::time_point start_time_;
+
+    double elapsed_seconds() const {
+        const auto elapsed = std::chrono::duration<double>(
+            std::chrono::steady_clock::now() - start_time_);
+        return std::max(elapsed.count(), 1e-9);
+    }
+
     std::string format_latency_summary(ylt::metric::histogram_t& hist) {
         // Access the internal sum and bucket counts
         auto sum_ptr =
@@ -274,6 +346,157 @@ struct MasterClientMetric {
     }
 };
 
+struct TransferOperationMetric {
+    std::array<std::string, 1> op_names = {"op_name"};
+
+    explicit TransferOperationMetric(
+        std::map<std::string, std::string> labels = {})
+        : read_op_count("mooncake_transfer_read_operation_count",
+                        "Total read operations by interface type", labels,
+                        op_names),
+          read_op_bytes("mooncake_transfer_read_operation_bytes",
+                        "Total read bytes by interface type", labels, op_names),
+          read_op_latency_us("mooncake_transfer_read_operation_latency",
+                             "Read operation latency by interface type (us)",
+                             kLatencyBucket, labels, op_names),
+          write_op_count("mooncake_transfer_write_operation_count",
+                         "Total write operations by interface type", labels,
+                         op_names),
+          write_op_bytes("mooncake_transfer_write_operation_bytes",
+                         "Total write bytes by interface type", labels,
+                         op_names),
+          write_op_latency_us("mooncake_transfer_write_operation_latency",
+                              "Write operation latency by interface type (us)",
+                              kLatencyBucket, labels, op_names) {}
+
+    ylt::metric::hybrid_counter_1t read_op_count;
+    ylt::metric::hybrid_counter_1t read_op_bytes;
+    ylt::metric::hybrid_histogram_1t read_op_latency_us;
+    ylt::metric::hybrid_counter_1t write_op_count;
+    ylt::metric::hybrid_counter_1t write_op_bytes;
+    ylt::metric::hybrid_histogram_1t write_op_latency_us;
+
+    void Observe(TransferOperationKind kind, const std::string& op_name,
+                 uint64_t bytes, uint64_t latency_us) {
+        const std::array<std::string, 1> label = {op_name};
+        {
+            std::lock_guard<std::mutex> lock(observed_ops_mutex_);
+            if (kind == TransferOperationKind::kRead) {
+                observed_read_ops_.insert(op_name);
+            } else {
+                observed_write_ops_.insert(op_name);
+            }
+        }
+
+        if (kind == TransferOperationKind::kRead) {
+            read_op_count.inc(label);
+            read_op_bytes.inc(label, bytes);
+            read_op_latency_us.observe(label, latency_us);
+        } else {
+            write_op_count.inc(label);
+            write_op_bytes.inc(label, bytes);
+            write_op_latency_us.observe(label, latency_us);
+        }
+    }
+
+    void serialize(std::string& str) {
+        read_op_count.serialize(str);
+        read_op_bytes.serialize(str);
+        read_op_latency_us.serialize(str);
+        write_op_count.serialize(str);
+        write_op_bytes.serialize(str);
+        write_op_latency_us.serialize(str);
+    }
+
+    std::string summary_metrics() {
+        std::stringstream ss;
+        ss << "=== Interface Operation Metrics Summary ===\n";
+        ss << format_operation_group_summary(
+                  "Read Interfaces", snapshot_operations(observed_read_ops_),
+                  read_op_count, read_op_bytes, read_op_latency_us)
+           << "\n";
+        ss << format_operation_group_summary(
+            "Write Interfaces", snapshot_operations(observed_write_ops_),
+            write_op_count, write_op_bytes, write_op_latency_us);
+        return ss.str();
+    }
+
+   private:
+    std::mutex observed_ops_mutex_;
+    std::unordered_set<std::string> observed_read_ops_;
+    std::unordered_set<std::string> observed_write_ops_;
+
+    std::vector<std::string> snapshot_operations(
+        const std::unordered_set<std::string>& source) {
+        std::lock_guard<std::mutex> lock(observed_ops_mutex_);
+        std::vector<std::string> ops(source.begin(), source.end());
+        std::sort(ops.begin(), ops.end());
+        return ops;
+    }
+
+    std::string format_operation_group_summary(
+        const std::string& group_name, const std::vector<std::string>& ops,
+        ylt::metric::hybrid_counter_1t& op_count,
+        ylt::metric::hybrid_counter_1t& op_bytes,
+        ylt::metric::hybrid_histogram_1t& op_latency_us) {
+        std::stringstream ss;
+        ss << group_name << ":\n";
+        if (ops.empty()) {
+            ss << "No data";
+            return ss.str();
+        }
+
+        auto bucket_counts = op_latency_us.get_bucket_counts();
+        bool found_any = false;
+        for (const auto& op_name : ops) {
+            const std::array<std::string, 1> label = {op_name};
+            const int64_t total_count = op_count.value(label);
+            if (total_count == 0) {
+                continue;
+            }
+
+            found_any = true;
+            ss << op_name << ": count=" << total_count << ", bytes="
+               << byte_size_to_string(
+                      static_cast<uint64_t>(op_bytes.value(label)));
+
+            int64_t p95_target = (total_count * 95) / 100;
+            int64_t cumulative = 0;
+            double p95_bucket = 0;
+            for (size_t i = 0;
+                 i < bucket_counts.size() && i < kLatencyBucket.size(); ++i) {
+                cumulative += bucket_counts[i]->value(label);
+                if (cumulative >= p95_target && p95_bucket == 0) {
+                    p95_bucket = kLatencyBucket[i];
+                    break;
+                }
+            }
+            if (p95_bucket > 0) {
+                ss << ", p95<" << p95_bucket << "μs";
+            }
+
+            double max_bucket = 0;
+            for (size_t i = bucket_counts.size(); i > 0; --i) {
+                const size_t idx = i - 1;
+                if (idx < kLatencyBucket.size() &&
+                    bucket_counts[idx]->value(label) > 0) {
+                    max_bucket = kLatencyBucket[idx];
+                    break;
+                }
+            }
+            if (max_bucket > 0) {
+                ss << ", max<" << max_bucket << "μs";
+            }
+            ss << "\n";
+        }
+
+        if (!found_any) {
+            ss << "No data";
+        }
+        return ss.str();
+    }
+};
+
 // SSD latency bucket: microseconds, tuned for SSD/network storage
 // Range: 50us (high-end NVMe) to 30s (3fs/nfs large object batch writes)
 inline const std::vector<double> kSsdLatencyBucket = {
@@ -424,7 +647,6 @@ struct SsdMetric {
         std::stringstream ss;
         ss << std::fixed << std::setprecision(1);
         ss << "count=" << count;
-        // rates order matches quantiles {0.5, 0.9, 0.99}
         if (rates.size() >= 1) ss << ", p50=" << rates[0] << "us";
         if (rates.size() >= 2) ss << ", p90=" << rates[1] << "us";
         if (rates.size() >= 3) ss << ", p99=" << rates[2] << "us";
@@ -438,6 +660,7 @@ struct SsdMetric {
 struct ClientMetric {
     TransferMetric transfer_metric;
     MasterClientMetric master_client_metric;
+    TransferOperationMetric transfer_operation_metric;
     SsdMetric ssd_metric;
 
     /**
@@ -452,25 +675,44 @@ struct ClientMetric {
      *   (default: 0, 0 = collect but don't report)
      */
     static std::unique_ptr<ClientMetric> Create(
-        const std::map<std::string, std::string>& labels = {});
+        const std::map<std::string, std::string>& labels = {},
+        bool master_rpc_metrics_enabled = true);
+
+    void ObserveTransferOperation(TransferOperationKind kind,
+                                  const std::string& op_name, uint64_t bytes,
+                                  uint64_t latency_us) {
+        transfer_operation_metric.Observe(kind, op_name, bytes, latency_us);
+    }
 
     void serialize(std::string& str);
     std::string summary_metrics();
 
     uint64_t GetReportingInterval() const { return metrics_interval_seconds_; }
 
-    explicit ClientMetric(
-        uint64_t interval_seconds = 0,
-        const std::map<std::string, std::string>& labels = {});
+    explicit ClientMetric(uint64_t interval_seconds = 0,
+                          const std::map<std::string, std::string>& labels = {},
+                          bool bandwidth_reporting_enabled = true,
+                          bool master_rpc_metrics_enabled = true);
     ~ClientMetric();
 
    private:
+    struct TransferSnapshot {
+        uint64_t read_bytes;
+        uint64_t write_bytes;
+        std::chrono::steady_clock::time_point timestamp;
+    };
+
     // Metrics reporting thread management
     std::jthread metrics_reporting_thread_;
     std::atomic<bool> should_stop_metrics_thread_{false};
     uint64_t metrics_interval_seconds_{0};
+    bool bandwidth_reporting_enabled_{true};
+    bool master_rpc_metrics_enabled_{true};
+    std::mutex snapshot_mutex_;
+    std::optional<TransferSnapshot> last_report_snapshot_;
 
     void StartMetricsReportingThread();
     void StopMetricsReportingThread();
+    std::string BuildBandwidthReport();
 };
-};  // namespace mooncake
+};  // namespace mooncake

mooncake-store/include/client_service.h

@@ -447,6 +447,15 @@ class Client {
         return master_client_.CalcCacheStats();
     }
 
+    void ObserveTransferOperation(TransferOperationKind kind,
+                                  const std::string& op_name, uint64_t bytes,
+                                  uint64_t latency_us) {
+        if (metrics_ != nullptr) {
+            metrics_->ObserveTransferOperation(kind, op_name, bytes,
+                                               latency_us);
+        }
+    }
+
     // For Prometheus-style metrics
     tl::expected<std::string, ErrorCode> SerializeMetrics() {
         if (metrics_ == nullptr) {

mooncake-store/include/dummy_client.h

@@ -7,6 +7,7 @@
 #include "pyclient.h"
 #include "real_client.h"
 #include "shm_helper.h"
+#include "client_metric.h"
 #include <memory>
 
 namespace mooncake {
@@ -203,6 +204,22 @@ class DummyClient : public PyClient {
     [[nodiscard]] std::vector<tl::expected<ResultType, ErrorCode>>
     invoke_batch_rpc(size_t input_size, Args &&...args);
 
+    template <auto ServiceMethod, typename... Args>
+    int invoke_observed_void_rpc(TransferOperationKind kind,
+                                 const char *op_name, size_t bytes, bool batch,
+                                 Args &&...args) {
+        auto result = execute_timed_operation<tl::expected<void, ErrorCode>>(
+            [&]() {
+                return invoke_rpc<ServiceMethod, void>(
+                    std::forward<Args>(args)...);
+            },
+            [](const auto &ret) { return ret.has_value(); },
+            [&](uint64_t latency_us, const auto &) {
+                ObserveTransferMetric(kind, op_name, bytes, latency_us, batch);
+            });
+        return to_py_ret(result);
+    }
+
     /**
      * @brief Accessor for the coro_rpc_client pool. Since coro_rpc_client
      * pool cannot reconnect to a different address, a new coro_rpc_client
@@ -261,6 +278,11 @@ class DummyClient : public PyClient {
 
     // Ascend physical device id for dummy-real RPC to real, set in setup_dummy
     int32_t device_id_ = 0;
+
+    std::unique_ptr<ClientMetric> metrics_;
+
+    void ObserveTransferMetric(TransferOperationKind kind, const char *op_name,
+                               size_t bytes, uint64_t latency_us, bool batch);
 };
 
 }  // namespace mooncake

mooncake-store/include/master_config.h

@@ -81,6 +81,10 @@ struct MasterConfig {
     std::string cxl_path;
     size_t cxl_size;
     bool enable_cxl = false;
+
+    // Offload-on-evict: defer LOCAL_DISK offload to eviction time
+    bool offload_on_evict = false;
+    bool offload_force_evict = false;
 };
 
 class MasterServiceSupervisorConfig {
@@ -140,6 +144,8 @@ class MasterServiceSupervisorConfig {
     std::string cxl_path = DEFAULT_CXL_PATH;
     size_t cxl_size = DEFAULT_CXL_SIZE;
     bool enable_cxl = false;
+    bool offload_on_evict = false;
+    bool offload_force_evict = false;
     MasterServiceSupervisorConfig() = default;
 
     // From MasterConfig
@@ -155,6 +161,8 @@ class MasterServiceSupervisorConfig {
         eviction_high_watermark_ratio = config.eviction_high_watermark_ratio;
         client_live_ttl_sec = config.client_live_ttl_sec;
         enable_offload = config.enable_offload;
+        offload_on_evict = config.offload_on_evict;
+        offload_force_evict = config.offload_force_evict;
         rpc_port = static_cast<int>(config.rpc_port);
         rpc_thread_num = static_cast<size_t>(config.rpc_thread_num);
 
@@ -267,6 +275,8 @@ class WrappedMasterServiceConfig {
     int64_t client_live_ttl_sec = DEFAULT_CLIENT_LIVE_TTL_SEC;
     bool enable_ha = false;
     bool enable_offload = false;
+    bool offload_on_evict = false;
+    bool offload_force_evict = false;
     std::string ha_backend_type = "etcd";
     std::string ha_backend_connstring;
     std::string cluster_id = DEFAULT_CLUSTER_ID;
@@ -322,6 +332,8 @@ class WrappedMasterServiceConfig {
         client_live_ttl_sec = config.client_live_ttl_sec;
         enable_ha = config.enable_ha;
         enable_offload = config.enable_offload;
+        offload_on_evict = config.offload_on_evict;
+        offload_force_evict = config.offload_force_evict;
         ha_backend_type = config.ha_backend_type;
         ha_backend_connstring = config.ha_backend_connstring;
         if (ha_backend_connstring.empty()) {
@@ -400,6 +412,8 @@ class WrappedMasterServiceConfig {
         enable_ha =
             true;  // This is used in HA mode, so enable_ha should be true
         enable_offload = config.enable_offload;
+        offload_on_evict = config.offload_on_evict;
+        offload_force_evict = config.offload_force_evict;
         ha_backend_type = config.ha_backend_type;
         ha_backend_connstring = config.ha_backend_connstring;
         if (ha_backend_connstring.empty()) {
@@ -736,6 +750,8 @@ class MasterServiceConfig {
     int64_t client_live_ttl_sec = DEFAULT_CLIENT_LIVE_TTL_SEC;
     bool enable_ha = false;
     bool enable_offload = false;
+    bool offload_on_evict = false;
+    bool offload_force_evict = false;
     std::string ha_backend_type = "etcd";
     std::string ha_backend_connstring;
     std::string cluster_id = DEFAULT_CLUSTER_ID;
@@ -787,6 +803,8 @@ class MasterServiceConfig {
         client_live_ttl_sec = config.client_live_ttl_sec;
         enable_ha = config.enable_ha;
         enable_offload = config.enable_offload;
+        offload_on_evict = config.offload_on_evict;
+        offload_force_evict = config.offload_force_evict;
         ha_backend_type = config.ha_backend_type;
         ha_backend_connstring = config.ha_backend_connstring;
         cluster_id = config.cluster_id;

mooncake-store/include/master_service.h

@@ -562,8 +562,14 @@ class MasterService {
     // fulfill evict ratio lowerbound.
     void BatchEvict(double evict_ratio_target, double evict_ratio_lowerbound);
 
+    // Helper to get a snapshot of alive clients (under client_mutex_ shared
+    // lock)
+    std::unordered_set<UUID, boost::hash<UUID>> getAliveClientsSnapshot() const;
+
     // Clear invalid handles in all shards
     void ClearInvalidHandles();
+    void ClearInvalidHandles(
+        const std::unordered_set<UUID, boost::hash<UUID>>& alive_clients);
 
     std::string FormatTimestamp(
         const std::chrono::system_clock::time_point& tp);
@@ -891,7 +897,10 @@ class MasterService {
     }
 
     // Helper to clean up stale handles pointing to unmounted segments
-    bool CleanupStaleHandles(ObjectMetadata& metadata);
+    // or local_disk replicas whose owner client is no longer alive.
+    bool CleanupStaleHandles(
+        ObjectMetadata& metadata,
+        const std::unordered_set<UUID, boost::hash<UUID>>& alive_clients);
 
     // Helper: allocate replicas, create ObjectMetadata, insert into shard,
     // and return descriptor list.  Shared by PutStart and UpsertStart.
@@ -962,11 +971,21 @@ class MasterService {
               it_(shard_guard_->metadata.find(key)),
               processing_it_(shard_guard_->processing_keys.find(key)),
               replication_task_it_(shard_guard_->replication_tasks.find(key)) {
-            // Automatically clean up invalid handles
+            // Automatically clean up invalid handles (memory replicas only).
+            // Note: We only check memory replicas here to avoid lock order
+            // violation (client_mutex_ must be acquired before metadata shard).
+            // local_disk replicas are cleaned up by ClearInvalidHandles() in
+            // ClientMonitorFunc.
             if (it_ != shard_guard_->metadata.end()) {
-                if (service_->CleanupStaleHandles(it_->second)) {
+                // Erase invalid memory replicas (those with unmounted
+                // segments). No client_mutex_ needed since we only check memory
+                // replicas.
+                it_->second.EraseReplicas([](const Replica& replica) {
+                    return replica.has_invalid_mem_handle();
+                });
+                // If no valid replicas remain, delete the whole object.
+                if (!it_->second.IsValid()) {
                     this->Erase();
-
                     if (processing_it_ != shard_guard_->processing_keys.end()) {
                         this->EraseFromProcessing();
                     }
@@ -1149,6 +1168,14 @@ class MasterService {
 
     const bool enable_offload_;
 
+    // Offload-on-evict: defer disk offload to eviction time
+    // (config: offload_on_evict)
+    bool offload_on_evict_{false};
+    // Force-evict: allow evicting MEMORY replicas without disk offload when cap
+    // exceeded (config: offload_force_evict, only effective when
+    // offload_on_evict_=true)
+    bool offload_force_evict_{false};
+
     const std::string ha_backend_type_;
 
     const std::string ha_backend_connstring_;

mooncake-store/include/replica.h

@@ -2,9 +2,12 @@
 
 #include <glog/logging.h>
 
+#include <boost/functional/hash.hpp>
+
 #include <cstdint>
 #include <memory>
 #include <string>
+#include <unordered_set>
 #include <variant>
 #include <vector>
 #include <unordered_map>
@@ -282,6 +285,36 @@ class Replica {
         return false;  // DiskReplicaData does not have handles
     }
 
+    /**
+     * @brief Check if a local_disk replica's owner client is still alive.
+     * Used by CleanupStaleHandles to remove replicas belonging to expired
+     * clients. For non-local_disk replicas, always returns false.
+     * @param alive_clients Set of currently alive client IDs.
+     * @return true if this is a local_disk replica whose client is not alive.
+     */
+    [[nodiscard]] bool has_stale_local_disk_client(
+        const std::unordered_set<UUID, boost::hash<UUID>>& alive_clients)
+        const {
+        auto client_id = get_local_disk_client_id();
+        if (client_id.has_value()) {
+            return alive_clients.find(client_id.value()) == alive_clients.end();
+        }
+        return false;
+    }
+
+    /**
+     * @brief Get the client_id for local_disk replicas.
+     * @return The client_id if this is a local_disk replica, std::nullopt
+     * otherwise.
+     */
+    [[nodiscard]] std::optional<UUID> get_local_disk_client_id() const {
+        if (is_local_disk_replica()) {
+            const auto& disk_data = std::get<LocalDiskReplicaData>(data_);
+            return disk_data.client_id;
+        }
+        return std::nullopt;
+    }
+
     [[nodiscard]] size_t get_memory_buffer_size() const {
         if (is_memory_replica()) {
             const auto& mem_data = std::get<MemoryReplicaData>(data_);

mooncake-store/include/segment.h

@@ -170,6 +170,12 @@ class ScopedSegmentAccess {
     ErrorCode SetSegmentStatusByName(const std::string& segment_name,
                                      SegmentStatus status);
 
+    /**
+     * @brief Remove the local disk segment entry for a client.
+     * Called when a client expires to clean up its local disk segment.
+     */
+    void UnmountLocalDiskSegment(const UUID& client_id);
+
    private:
     SegmentManager* segment_manager_;
     std::unique_lock<std::shared_mutex> lock_;

mooncake-store/src/client_metric.cpp

@@ -28,6 +28,27 @@ bool parseMetricsEnabled() {
             value == "on" || value == "enable");
 }
 
+bool parseBoolEnv(const char* env_name, bool default_value) {
+    const char* env_value = std::getenv(env_name);
+    if (!env_value) {
+        return default_value;
+    }
+
+    std::string value = toLower(env_value);
+    if (value == "1" || value == "true" || value == "yes" || value == "on" ||
+        value == "enable") {
+        return true;
+    }
+    if (value == "0" || value == "false" || value == "no" || value == "off" ||
+        value == "disable") {
+        return false;
+    }
+
+    LOG(WARNING) << "Failed to parse " << env_name << ": " << env_value
+                 << ", fallback to default=" << default_value;
+    return default_value;
+}
+
 uint64_t parseMetricsInterval() {
     const char* interval_env = std::getenv("MC_STORE_CLIENT_METRIC_INTERVAL");
     if (!interval_env) {
@@ -55,12 +76,21 @@ uint64_t parseMetricsInterval() {
 }  // anonymous namespace
 
 ClientMetric::ClientMetric(uint64_t interval_seconds,
-                           const std::map<std::string, std::string>& labels)
+                           const std::map<std::string, std::string>& labels,
+                           bool bandwidth_reporting_enabled,
+                           bool master_rpc_metrics_enabled)
     : transfer_metric(labels),
       master_client_metric(labels),
+      transfer_operation_metric(labels),
       ssd_metric(labels),
       should_stop_metrics_thread_(false),
-      metrics_interval_seconds_(interval_seconds) {
+      metrics_interval_seconds_(interval_seconds),
+      bandwidth_reporting_enabled_(bandwidth_reporting_enabled),
+      master_rpc_metrics_enabled_(master_rpc_metrics_enabled) {
+    last_report_snapshot_ = TransferSnapshot{
+        static_cast<uint64_t>(transfer_metric.total_read_bytes.value()),
+        static_cast<uint64_t>(transfer_metric.total_write_bytes.value()),
+        std::chrono::steady_clock::now()};
     if (metrics_interval_seconds_ > 0) {
         StartMetricsReportingThread();
     }
@@ -69,7 +99,8 @@ ClientMetric::ClientMetric(uint64_t interval_seconds,
 ClientMetric::~ClientMetric() { StopMetricsReportingThread(); }
 
 std::unique_ptr<ClientMetric> ClientMetric::Create(
-    const std::map<std::string, std::string>& labels) {
+    const std::map<std::string, std::string>& labels,
+    bool master_rpc_metrics_enabled) {
     if (!parseMetricsEnabled()) {
         LOG(INFO) << "Client metrics disabled (set MC_STORE_CLIENT_METRIC=0 to "
                      "disable)";
@@ -77,29 +108,83 @@ std::unique_ptr<ClientMetric> ClientMetric::Create(
     }
 
     uint64_t interval = parseMetricsInterval();
+    bool bandwidth_reporting_enabled =
+        parseBoolEnv("MC_STORE_CLIENT_METRIC_BANDWIDTH", true);
 
     LOG(INFO) << "Client metrics enabled (default enabled)";
+    LOG(INFO) << "Client bandwidth summary "
+              << (bandwidth_reporting_enabled ? "enabled" : "disabled")
+              << " via MC_STORE_CLIENT_METRIC_BANDWIDTH";
 
-    return std::make_unique<ClientMetric>(interval, labels);
+    return std::make_unique<ClientMetric>(interval, labels,
+                                          bandwidth_reporting_enabled,
+                                          master_rpc_metrics_enabled);
 }
 
 void ClientMetric::serialize(std::string& str) {
     transfer_metric.serialize(str);
-    master_client_metric.serialize(str);
+    if (master_rpc_metrics_enabled_) {
+        master_client_metric.serialize(str);
+    }
+    transfer_operation_metric.serialize(str);
     ssd_metric.serialize(str);
 }
 
 std::string ClientMetric::summary_metrics() {
     std::stringstream ss;
     ss << "Client Metrics Summary\n";
-    ss << transfer_metric.summary_metrics();
+    ss << transfer_metric.summary_metrics(bandwidth_reporting_enabled_);
     ss << "\n";
-    ss << master_client_metric.summary_metrics();
+    if (master_rpc_metrics_enabled_) {
+        ss << master_client_metric.summary_metrics();
+        ss << "\n";
+    }
+    ss << transfer_operation_metric.summary_metrics();
     ss << "\n";
     ss << ssd_metric.summary_metrics();
     return ss.str();
 }
 
+std::string ClientMetric::BuildBandwidthReport() {
+    if (!bandwidth_reporting_enabled_) {
+        return "";
+    }
+
+    const auto now = std::chrono::steady_clock::now();
+    const uint64_t read_bytes = transfer_metric.total_read_bytes.value();
+    const uint64_t write_bytes = transfer_metric.total_write_bytes.value();
+
+    std::lock_guard<std::mutex> lock(snapshot_mutex_);
+    if (!last_report_snapshot_.has_value()) {
+        last_report_snapshot_ = TransferSnapshot{read_bytes, write_bytes, now};
+        return "";
+    }
+
+    const auto previous = *last_report_snapshot_;
+    last_report_snapshot_ = TransferSnapshot{read_bytes, write_bytes, now};
+
+    const double elapsed_seconds = std::max(
+        std::chrono::duration<double>(now - previous.timestamp).count(), 1e-9);
+    const uint64_t read_delta = read_bytes >= previous.read_bytes
+                                    ? read_bytes - previous.read_bytes
+                                    : 0;
+    const uint64_t write_delta = write_bytes >= previous.write_bytes
+                                     ? write_bytes - previous.write_bytes
+                                     : 0;
+
+    std::stringstream ss;
+    ss << "=== Interval Throughput Summary ===\n";
+    ss << "Read Throughput: "
+       << format_metric_rate(read_delta / elapsed_seconds, "B/s") << " ("
+       << byte_size_to_string(read_delta) << " over " << std::fixed
+       << std::setprecision(2) << elapsed_seconds << "s)\n";
+    ss << "Write Throughput: "
+       << format_metric_rate(write_delta / elapsed_seconds, "B/s") << " ("
+       << byte_size_to_string(write_delta) << " over " << std::fixed
+       << std::setprecision(2) << elapsed_seconds << "s)";
+    return ss.str();
+}
+
 void ClientMetric::StartMetricsReportingThread() {
     should_stop_metrics_thread_ = false;
     metrics_reporting_thread_ =
@@ -124,7 +209,12 @@ void ClientMetric::StartMetricsReportingThread() {
 
                 // Print metrics summary
                 std::string summary = summary_metrics();
-                LOG(INFO) << "Client Metrics Report:\n" << summary;
+                std::string bandwidth_report = BuildBandwidthReport();
+                std::string report = "Client Metrics Report:\n" + summary;
+                if (!bandwidth_report.empty()) {
+                    report += "\n" + bandwidth_report;
+                }
+                LOG(INFO) << report;
             }
             LOG(INFO) << "Client metrics reporting thread stopped";
         });
@@ -140,4 +230,4 @@ void ClientMetric::StopMetricsReportingThread() {
     }
 }
 
-}  // namespace mooncake
+}  // namespace mooncake

mooncake-store/src/dummy_client.cpp

@@ -7,6 +7,7 @@
 #include <sys/stat.h>  // For S_IRUSR, S_IWUSR
 #include <fcntl.h>     // For O_CREAT, O_RDWR
 #include <unistd.h>    // For ftruncate, close, shm_unlink
+#include <chrono>
 #include <cstdlib>
 
 #include "real_client.h"
@@ -23,6 +24,75 @@
 #endif
 
 namespace {
+size_t sum_value_sizes(const std::vector<std::span<const char>>& values) {
+    size_t total = 0;
+    for (const auto& value : values) {
+        total += value.size_bytes();
+    }
+    return total;
+}
+
+size_t sum_sizes(const std::vector<size_t>& sizes) {
+    size_t total = 0;
+    for (size_t size : sizes) {
+        total += size;
+    }
+    return total;
+}
+
+size_t sum_successful_sizes(const std::vector<int>& results,
+                            const std::vector<size_t>& sizes) {
+    size_t total = 0;
+    for (size_t i = 0; i < results.size() && i < sizes.size(); ++i) {
+        if (results[i] == 0) {
+            total += sizes[i];
+        }
+    }
+    return total;
+}
+
+size_t sum_successful_nested_sizes(
+    const std::vector<int>& results,
+    const std::vector<std::vector<size_t>>& nested_sizes) {
+    size_t total = 0;
+    for (size_t i = 0; i < results.size() && i < nested_sizes.size(); ++i) {
+        if (results[i] == 0) {
+            total += sum_sizes(nested_sizes[i]);
+        }
+    }
+    return total;
+}
+
+size_t sum_positive_results(const std::vector<int64_t>& results) {
+    size_t total = 0;
+    for (int64_t result : results) {
+        if (result > 0) {
+            total += static_cast<size_t>(result);
+        }
+    }
+    return total;
+}
+
+size_t sum_positive_results(const std::vector<int>& results) {
+    size_t total = 0;
+    for (int result : results) {
+        if (result > 0) {
+            total += static_cast<size_t>(result);
+        }
+    }
+    return total;
+}
+
+size_t sum_positive_ranges(
+    const std::vector<std::vector<std::vector<int64_t>>>& results) {
+    size_t total = 0;
+    for (const auto& key_rows : results) {
+        for (const auto& row : key_rows) {
+            total += sum_positive_results(row);
+        }
+    }
+    return total;
+}
 
 std::vector<uint64_t> void_ptrs_to_u64(const std::vector<void*>& ptrs) {
     std::vector<uint64_t> out;
@@ -129,7 +199,10 @@ std::vector<tl::expected<ResultType, ErrorCode>> DummyClient::invoke_batch_rpc(
         }());
 }
 
-DummyClient::DummyClient() : client_id_(generate_uuid()) {
+DummyClient::DummyClient()
+    : client_id_(generate_uuid()),
+      metrics_(ClientMetric::Create(merge_labels({{"client_mode", "dummy"}}),
+                                    false)) {
     // Initialize logging severity (leave as before)
     mooncake::init_ylt_log_level();
     // Initialize client pools
@@ -141,6 +214,30 @@ DummyClient::DummyClient() : client_id_(generate_uuid()) {
 
 DummyClient::~DummyClient() { tearDownAll(); }
 
+void DummyClient::ObserveTransferMetric(TransferOperationKind kind,
+                                        const char* op_name, size_t bytes,
+                                        uint64_t latency_us, bool batch) {
+    if (!metrics_) {
+        return;
+    }
+    metrics_->ObserveTransferOperation(kind, op_name, bytes, latency_us);
+    if (kind == TransferOperationKind::kRead) {
+        metrics_->transfer_metric.total_read_bytes.inc(bytes);
+        if (batch) {
+            metrics_->transfer_metric.batch_get_latency_us.observe(latency_us);
+        } else {
+            metrics_->transfer_metric.get_latency_us.observe(latency_us);
+        }
+    } else {
+        metrics_->transfer_metric.total_write_bytes.inc(bytes);
+        if (batch) {
+            metrics_->transfer_metric.batch_put_latency_us.observe(latency_us);
+        } else {
+            metrics_->transfer_metric.put_latency_us.observe(latency_us);
+        }
+    }
+}
+
 ErrorCode DummyClient::connect(const std::string& server_address) {
     ScopedVLogTimer timer(1, "DummyClient::Connect");
     timer.LogRequest("real_client_addr=", server_address);
@@ -524,35 +621,40 @@ uint64_t DummyClient::alloc_from_mem_pool(size_t size) {
 
 int DummyClient::put(const std::string& key, std::span<const char> value,
                      const ReplicateConfig& config) {
-    return to_py_ret(invoke_rpc<&RealClient::put_dummy_helper, void>(
-        key, value, config, client_id_));
+    return invoke_observed_void_rpc<&RealClient::put_dummy_helper>(
+        TransferOperationKind::kWrite, "put", value.size_bytes(), false, key,
+        value, config, client_id_);
 }
 
 int DummyClient::put_batch(const std::vector<std::string>& keys,
                            const std::vector<std::span<const char>>& values,
                            const ReplicateConfig& config) {
-    return to_py_ret(invoke_rpc<&RealClient::put_batch_dummy_helper, void>(
-        keys, values, config, client_id_));
+    return invoke_observed_void_rpc<&RealClient::put_batch_dummy_helper>(
+        TransferOperationKind::kWrite, "put_batch", sum_value_sizes(values),
+        true, keys, values, config, client_id_);
 }
 
 int DummyClient::put_parts(const std::string& key,
                            std::vector<std::span<const char>> values,
                            const ReplicateConfig& config) {
-    return to_py_ret(invoke_rpc<&RealClient::put_parts_dummy_helper, void>(
-        key, values, config, client_id_));
+    return invoke_observed_void_rpc<&RealClient::put_parts_dummy_helper>(
+        TransferOperationKind::kWrite, "put_parts", sum_value_sizes(values),
+        false, key, values, config, client_id_);
 }
 
 int DummyClient::upsert(const std::string& key, std::span<const char> value,
                         const ReplicateConfig& config) {
-    return to_py_ret(invoke_rpc<&RealClient::upsert_dummy_helper, void>(
-        key, value, config, client_id_));
+    return invoke_observed_void_rpc<&RealClient::upsert_dummy_helper>(
+        TransferOperationKind::kWrite, "upsert", value.size_bytes(), false, key,
+        value, config, client_id_);
 }
 
 int DummyClient::upsert_from(const std::string& key, void* buffer, size_t size,
                              const ReplicateConfig& config) {
     uint64_t dummy_addr = reinterpret_cast<uint64_t>(buffer);
-    return to_py_ret(invoke_rpc<&RealClient::upsert_from_dummy_helper, void>(
-        key, dummy_addr, size, config, client_id_));
+    return invoke_observed_void_rpc<&RealClient::upsert_from_dummy_helper>(
+        TransferOperationKind::kWrite, "upsert_from", size, false, key,
+        dummy_addr, size, config, client_id_);
 }
 
 std::vector<int> DummyClient::batch_upsert_from(
@@ -562,6 +664,7 @@ std::vector<int> DummyClient::batch_upsert_from(
     for (auto ptr : buffer_ptrs) {
         buffers.push_back(reinterpret_cast<uint64_t>(ptr));
     }
+    const auto start_time = std::chrono::steady_clock::now();
     auto internal_results =
         invoke_batch_rpc<&RealClient::batch_upsert_from_dummy_helper, void>(
             keys.size(), keys, buffers, sizes, config, client_id_);
@@ -570,21 +673,29 @@ std::vector<int> DummyClient::batch_upsert_from(
     for (const auto& result : internal_results) {
         results.push_back(to_py_ret(result));
     }
+    const size_t successful_bytes = sum_successful_sizes(results, sizes);
+    if (successful_bytes > 0) {
+        ObserveTransferMetric(TransferOperationKind::kWrite,
+                              "batch_upsert_from", successful_bytes,
+                              elapsed_us_since(start_time), true);
+    }
     return results;
 }
 
 int DummyClient::upsert_parts(const std::string& key,
                               std::vector<std::span<const char>> values,
                               const ReplicateConfig& config) {
-    return to_py_ret(invoke_rpc<&RealClient::upsert_parts_dummy_helper, void>(
-        key, values, config, client_id_));
+    return invoke_observed_void_rpc<&RealClient::upsert_parts_dummy_helper>(
+        TransferOperationKind::kWrite, "upsert_parts", sum_value_sizes(values),
+        false, key, values, config, client_id_);
 }
 
 int DummyClient::upsert_batch(const std::vector<std::string>& keys,
                               const std::vector<std::span<const char>>& values,
                               const ReplicateConfig& config) {
-    return to_py_ret(invoke_rpc<&RealClient::upsert_batch_dummy_helper, void>(
-        keys, values, config, client_id_));
+    return invoke_observed_void_rpc<&RealClient::upsert_batch_dummy_helper>(
+        TransferOperationKind::kWrite, "upsert_batch", sum_value_sizes(values),
+        true, keys, values, config, client_id_);
 }
 
 int DummyClient::remove(const std::string& key, bool force) {
@@ -650,6 +761,7 @@ int64_t DummyClient::getSize(const std::string& key) {
 }
 
 std::shared_ptr<BufferHandle> DummyClient::get_buffer(const std::string& key) {
+    const auto start_time = std::chrono::steady_clock::now();
     // Try hot cache path if shm is mapped
     if (hot_cache_base_) {
         auto result = invoke_rpc<&RealClient::acquire_hot_cache,
@@ -668,6 +780,8 @@ std::shared_ptr<BufferHandle> DummyClient::get_buffer(const std::string& key) {
                 (void)invoke_rpc<&RealClient::release_hot_cache, void>(
                     key_copy);
             };
+            ObserveTransferMetric(TransferOperationKind::kRead, "get_buffer",
+                                  size, elapsed_us_since(start_time), false);
             return std::make_shared<BufferHandle>(local_ptr, size,
                                                   std::move(release));
         }
@@ -686,11 +800,14 @@ std::shared_ptr<BufferHandle> DummyClient::get_buffer(const std::string& key) {
         (void)invoke_rpc<&RealClient::release_buffer_dummy, void>(dummy_addr,
                                                                   client_id_);
     };
+    ObserveTransferMetric(TransferOperationKind::kRead, "get_buffer", size,
+                          elapsed_us_since(start_time), false);
     return std::make_shared<BufferHandle>(local_ptr, size, std::move(release));
 }
 
 std::vector<std::shared_ptr<BufferHandle>> DummyClient::batch_get_buffer(
     const std::vector<std::string>& keys) {
+    const auto start_time = std::chrono::steady_clock::now();
     std::vector<std::shared_ptr<BufferHandle>> results(keys.size(), nullptr);
     if (keys.empty()) return results;
 
@@ -751,19 +868,37 @@ std::vector<std::shared_ptr<BufferHandle>> DummyClient::batch_get_buffer(
             std::make_shared<BufferHandle>(ptr, size, std::move(release));
     }
 
+    size_t total_bytes = 0;
+    for (const auto& result : results) {
+        if (result != nullptr) {
+            total_bytes += result->size();
+        }
+    }
+    if (total_bytes > 0) {
+        ObserveTransferMetric(TransferOperationKind::kRead, "batch_get_buffer",
+                              total_bytes, elapsed_us_since(start_time), true);
+    }
+
     return results;
 }
 
 int64_t DummyClient::get_into(const std::string& key, void* buffer,
                               size_t size) {
     uint64_t buf_addr = reinterpret_cast<uint64_t>(buffer);
+    const auto start_time = std::chrono::steady_clock::now();
     auto result = invoke_rpc<&RealClient::get_into_range_shm_helper,
                              tl::expected<int64_t, ErrorCode>>(
         key, buf_addr, 0, 0, size, client_id_);
     if (!result) {
         return static_cast<int64_t>(toInt(result.error()));
     }
-    return to_py_ret(*result);
+    const int64_t bytes_read = to_py_ret(*result);
+    if (bytes_read >= 0) {
+        ObserveTransferMetric(TransferOperationKind::kRead, "get_into",
+                              static_cast<size_t>(bytes_read),
+                              elapsed_us_since(start_time), false);
+    }
+    return bytes_read;
 }
 
 std::vector<std::vector<std::vector<int64_t>>> DummyClient::get_into_ranges(
@@ -773,6 +908,7 @@ std::vector<std::vector<std::vector<int64_t>>> DummyClient::get_into_ranges(
     const std::vector<std::vector<std::vector<size_t>>>& all_src_offsets,
     const std::vector<std::vector<std::vector<size_t>>>& all_sizes) {
     std::vector<uint64_t> dummy_buffers = void_ptrs_to_u64(buffers);
+    const auto start_time = std::chrono::steady_clock::now();
     auto internal_results =
         invoke_rpc<&RealClient::get_into_ranges_shm_helper,
                    std::vector<std::vector<
@@ -786,8 +922,13 @@ std::vector<std::vector<std::vector<int64_t>>> DummyClient::get_into_ranges(
                                                all_dst_offsets,
                                                internal_results.error());
     }
-
-    return convert_ranged_read_results(internal_results.value());
+    auto results = convert_ranged_read_results(internal_results.value());
+    const size_t total_bytes = sum_positive_ranges(results);
+    if (total_bytes > 0) {
+        ObserveTransferMetric(TransferOperationKind::kRead, "get_into_ranges",
+                              total_bytes, elapsed_us_since(start_time), true);
+    }
+    return results;
 }
 
 std::string DummyClient::get_hostname() const {
@@ -799,6 +940,7 @@ std::vector<int> DummyClient::batch_put_from(
     const std::vector<std::string>& keys, const std::vector<void*>& buffer_ptrs,
     const std::vector<size_t>& sizes, const ReplicateConfig& config) {
     std::vector<uint64_t> buffers = void_ptrs_to_u64(buffer_ptrs);
+    const auto start_time = std::chrono::steady_clock::now();
     auto internal_results =
         invoke_batch_rpc<&RealClient::batch_put_from_dummy_helper, void>(
             keys.size(), keys, buffers, sizes, config, device_id_, client_id_);
@@ -809,6 +951,13 @@ std::vector<int> DummyClient::batch_put_from(
         results.push_back(to_py_ret(result));
     }
 
+    const size_t successful_bytes = sum_successful_sizes(results, sizes);
+    if (successful_bytes > 0) {
+        ObserveTransferMetric(TransferOperationKind::kWrite, "batch_put_from",
+                              successful_bytes, elapsed_us_since(start_time),
+                              true);
+    }
+
     return results;
 }
 
@@ -822,6 +971,7 @@ std::vector<int64_t> DummyClient::batch_get_into(
     const std::vector<std::string>& keys, const std::vector<void*>& buffer_ptrs,
     const std::vector<size_t>& sizes) {
     std::vector<uint64_t> buffers = void_ptrs_to_u64(buffer_ptrs);
+    const auto start_time = std::chrono::steady_clock::now();
     auto internal_results =
         invoke_batch_rpc<&RealClient::batch_get_into_dummy_helper, int64_t>(
             keys.size(), keys, buffers, sizes, device_id_, client_id_);
@@ -832,6 +982,12 @@ std::vector<int64_t> DummyClient::batch_get_into(
         results.push_back(to_py_ret(result));
     }
 
+    const size_t total_bytes = sum_positive_results(results);
+    if (total_bytes > 0) {
+        ObserveTransferMetric(TransferOperationKind::kRead, "batch_get_into",
+                              total_bytes, elapsed_us_since(start_time), true);
+    }
+
     return results;
 }
 
@@ -850,6 +1006,7 @@ std::vector<int> DummyClient::batch_put_from_multi_buffers(
     const ReplicateConfig& config) {
     std::vector<std::vector<uint64_t>> dummy_nested =
         void_ptr_rows_to_u64_nested(all_buffer_ptrs);
+    const auto start_time = std::chrono::steady_clock::now();
     auto internal_results =
         invoke_batch_rpc<&RealClient::batch_put_from_multi_buffers_dummy_helper,
                          void>(keys.size(), keys, dummy_nested, all_sizes,
@@ -859,6 +1016,13 @@ std::vector<int> DummyClient::batch_put_from_multi_buffers(
     for (const auto& result : internal_results) {
         results.push_back(to_py_ret(result));
     }
+    const size_t successful_bytes =
+        sum_successful_nested_sizes(results, all_sizes);
+    if (successful_bytes > 0) {
+        ObserveTransferMetric(TransferOperationKind::kWrite,
+                              "batch_put_from_multi_buffers", successful_bytes,
+                              elapsed_us_since(start_time), true);
+    }
     return results;
 }
 
@@ -869,6 +1033,7 @@ std::vector<int> DummyClient::batch_get_into_multi_buffers(
     bool prefer_alloc_in_same_node) {
     std::vector<std::vector<uint64_t>> dummy_nested =
         void_ptr_rows_to_u64_nested(all_buffer_ptrs);
+    const auto start_time = std::chrono::steady_clock::now();
     auto internal_results =
         invoke_batch_rpc<&RealClient::batch_get_into_multi_buffers_dummy_helper,
                          int64_t>(keys.size(), keys, dummy_nested, all_sizes,
@@ -879,6 +1044,12 @@ std::vector<int> DummyClient::batch_get_into_multi_buffers(
     for (const auto& result : internal_results) {
         results.push_back(to_py_ret(result));
     }
+    const size_t total_bytes = sum_positive_results(results);
+    if (total_bytes > 0) {
+        ObserveTransferMetric(TransferOperationKind::kRead,
+                              "batch_get_into_multi_buffers", total_bytes,
+                              elapsed_us_since(start_time), true);
+    }
     return results;
 }
 

mooncake-store/src/file_storage.cpp

@@ -42,9 +42,10 @@ FileStorageConfig FileStorageConfig::FromEnvironment() {
     config.local_buffer_size = GetEnvOr<int64_t>(
         "MOONCAKE_OFFLOAD_LOCAL_BUFFER_SIZE_BYTES", config.local_buffer_size);
 
-    config.scanmeta_iterator_keys_limit =
+    config.scanmeta_iterator_keys_limit = GetEnvOr<int64_t>(
+        "MOONCAKE_OFFLOAD_SCANMETA_ITERATOR_KEYS_LIMIT",
         GetEnvOr<int64_t>("MOONCAKE_SCANMETA_ITERATOR_KEYS_LIMIT",
-                          config.scanmeta_iterator_keys_limit);
+                          config.scanmeta_iterator_keys_limit));
 
     config.total_keys_limit = GetEnvOr<int64_t>(
         "MOONCAKE_OFFLOAD_TOTAL_KEYS_LIMIT", config.total_keys_limit);
@@ -63,7 +64,9 @@ FileStorageConfig FileStorageConfig::FromEnvironment() {
         GetEnvOr<uint64_t>("MOONCAKE_OFFLOAD_CLIENT_BUFFER_GC_TTL_MS",
                            config.client_buffer_gc_ttl_ms);
 
-    auto use_uring_str = GetEnvStringOr("MOONCAKE_USE_URING", "false");
+    auto use_uring_str =
+        GetEnvStringOr("MOONCAKE_OFFLOAD_USE_URING",
+                       GetEnvStringOr("MOONCAKE_USE_URING", "false"));
     config.use_uring = (use_uring_str == "true" || use_uring_str == "1");
 
     return config;

mooncake-store/src/master.cpp

@@ -109,6 +109,10 @@ DEFINE_validator(eviction_ratio, [](const char* flagname, double value) {
 DEFINE_bool(enable_ha, false,
             "Enable high availability, which depends on etcd");
 DEFINE_bool(enable_offload, false, "Enable offload availability");
+DEFINE_bool(offload_on_evict, false,
+            "Defer LOCAL_DISK offload to eviction time instead of PutEnd");
+DEFINE_bool(offload_force_evict, false,
+            "Force-evict objects exceeding offload cap without disk offload");
 DEFINE_string(ha_backend_type, "etcd",
               "HA backend type, e.g. etcd | redis | k8s");
 DEFINE_string(ha_backend_connstring, "",
@@ -293,6 +297,11 @@ void InitMasterConf(const mooncake::DefaultConfig& default_config,
                            FLAGS_enable_ha);
     default_config.GetBool("enable_offload", &master_config.enable_offload,
                            FLAGS_enable_offload);
+    default_config.GetBool("offload_on_evict", &master_config.offload_on_evict,
+                           FLAGS_offload_on_evict);
+    default_config.GetBool("offload_force_evict",
+                           &master_config.offload_force_evict,
+                           FLAGS_offload_force_evict);
     default_config.GetString("ha_backend_type", &master_config.ha_backend_type,
                              FLAGS_ha_backend_type);
     default_config.GetString("ha_backend_connstring",
@@ -867,6 +876,8 @@ int main(int argc, char* argv[]) {
         << master_config.eviction_high_watermark_ratio
         << ", enable_ha=" << master_config.enable_ha
         << ", enable_offload=" << master_config.enable_offload
+        << ", offload_on_evict=" << master_config.offload_on_evict
+        << ", offload_force_evict=" << master_config.offload_force_evict
         << ", ha_backend_type=" << master_config.ha_backend_type
         << ", ha_backend_connstring=" << ha_backend_connstring
         << ", etcd_endpoints=" << master_config.etcd_endpoints

mooncake-store/src/master_service.cpp

@@ -48,6 +48,13 @@ namespace {
 
 constexpr size_t kUnlimitedSnapshotList = 0;
 
+// Per-cycle offload cap as a fraction of `offloading_queue_limit_`. Used only
+// when offload-on-evict mode is active. Defers memory eviction for at most
+// this fraction of the queue limit per BatchEvict cycle; beyond that, eviction
+// falls back according to `offload_force_evict_`. A future change may expose
+// this as a configurable parameter if workloads demand tuning.
+constexpr double kOffloadCapRatio = 0.5;
+
 enum class SnapshotCatalogBackendKind {
     kEmbedded,
     kRedis,
@@ -159,6 +166,19 @@ MasterService::MasterService(const MasterServiceConfig& config)
             "put_start_discard_timeout_sec");
     }
 
+    // Offload-on-evict: defer LOCAL_DISK offload to eviction time
+    offload_on_evict_ = enable_offload_ && config.offload_on_evict;
+    if (offload_on_evict_) {
+        LOG(INFO) << "Offload-on-evict mode enabled: DRAM offload to "
+                     "LOCAL_DISK will occur at eviction time instead of "
+                     "PutEnd";
+        offload_force_evict_ = config.offload_force_evict;
+        if (offload_force_evict_) {
+            LOG(INFO) << "Force-evict enabled: objects exceeding offload "
+                         "cap will be evicted without disk offload";
+        }
+    }
+
     eviction_running_ = true;
     eviction_thread_ = std::thread(&MasterService::EvictionThreadFunc, this);
     VLOG(1) << "action=start_eviction_thread";
@@ -351,12 +371,23 @@ auto MasterService::ReMountSegment(const std::vector<Segment>& segments,
     return {};
 }
 
+std::unordered_set<UUID, boost::hash<UUID>>
+MasterService::getAliveClientsSnapshot() const {
+    std::shared_lock<std::shared_mutex> lock(client_mutex_);
+    return ok_client_;
+}
+
 void MasterService::ClearInvalidHandles() {
+    ClearInvalidHandles(getAliveClientsSnapshot());
+}
+
+void MasterService::ClearInvalidHandles(
+    const std::unordered_set<UUID, boost::hash<UUID>>& alive_clients) {
     for (size_t i = 0; i < kNumShards; i++) {
         MetadataShardAccessorRW shard(this, i);
         auto it = shard->metadata.begin();
         while (it != shard->metadata.end()) {
-            if (CleanupStaleHandles(it->second)) {
+            if (CleanupStaleHandles(it->second, alive_clients)) {
                 // If the object is empty, we need to erase the iterator and
                 // also erase the key from processing_keys,
                 // replication_tasks, and offloading_tasks.
@@ -842,13 +873,15 @@ auto MasterService::PutStart(const UUID& client_id, const std::string& key,
     VLOG(1) << "key=" << key << ", value_length=" << slice_length
             << ", config=" << config << ", action=put_start_begin";
 
+    auto alive_clients = getAliveClientsSnapshot();
     std::shared_lock<std::shared_mutex> shared_lock(snapshot_mutex_);
     // Lock the shard and check if object already exists
     MetadataShardAccessorRW shard(this, getShardIndex(key));
 
     const auto now = std::chrono::system_clock::now();
     auto it = shard->metadata.find(key);
-    if (it != shard->metadata.end() && !CleanupStaleHandles(it->second)) {
+    if (it != shard->metadata.end() &&
+        !CleanupStaleHandles(it->second, alive_clients)) {
         auto& metadata = it->second;
         // If the object's PutStart expired and has not completed any
         // replicas, we can discard it and allow the new PutStart to
@@ -897,7 +930,7 @@ auto MasterService::PutEnd(const UUID& client_id, const std::string& key,
         },
         [](Replica& replica) { replica.mark_complete(); });
 
-    if (enable_offload_) {
+    if (enable_offload_ && !offload_on_evict_) {
         auto& shard = accessor.GetShard();
         metadata.VisitReplicas(
             &Replica::fn_is_completed, [this, &key, &shard](Replica& replica) {
@@ -1085,6 +1118,7 @@ auto MasterService::UpsertStart(const UUID& client_id, const std::string& key,
     //   during full metadata snapshots.
     // shard lock (exclusive via MetadataShardAccessorRW): serializes all
     //   operations on keys that hash to the same shard.
+    auto alive_clients = getAliveClientsSnapshot();
     std::shared_lock<std::shared_mutex> shared_lock(snapshot_mutex_);
     MetadataShardAccessorRW shard(this, getShardIndex(key));
 
@@ -1094,7 +1128,9 @@ auto MasterService::UpsertStart(const UUID& client_id, const std::string& key,
     // --- Step 0: stale handle cleanup ---
     // If all memory replicas point to unmounted segments (node crashed and
     // restarted), the metadata is useless — erase it and treat as new key.
-    if (it != shard->metadata.end() && CleanupStaleHandles(it->second)) {
+    // Also clean up local_disk replicas whose owner client has expired.
+    if (it != shard->metadata.end() &&
+        CleanupStaleHandles(it->second, alive_clients)) {
         shard->processing_keys.erase(key);
         shard->metadata.erase(it);
         it = shard->metadata.end();
@@ -1931,6 +1967,8 @@ auto MasterService::BatchRemove(const std::vector<std::string>& keys,
 
     std::shared_lock<std::shared_mutex> snapshot_lock(snapshot_mutex_);
 
+    auto alive_clients = getAliveClientsSnapshot();
+
     // Process each shard once, acquiring lock per shard
     for (auto& [shard_idx, key_group] : keys_by_shard) {
         MetadataShardAccessorRW shard(this, shard_idx);
@@ -1948,7 +1986,7 @@ auto MasterService::BatchRemove(const std::vector<std::string>& keys,
             }
 
             // Clean up stale replica handles (consistent with single Remove)
-            if (CleanupStaleHandles(it->second)) {
+            if (CleanupStaleHandles(it->second, alive_clients)) {
                 shard->processing_keys.erase(key);
                 shard->replication_tasks.erase(key);
                 shard->offloading_tasks.erase(key);
@@ -1996,10 +2034,14 @@ auto MasterService::BatchRemove(const std::vector<std::string>& keys,
     return results;
 }
 
-bool MasterService::CleanupStaleHandles(ObjectMetadata& metadata) {
-    // Remove those with invalid allocators
-    metadata.EraseReplicas([](const Replica& replica) {
-        return replica.has_invalid_mem_handle();
+bool MasterService::CleanupStaleHandles(
+    ObjectMetadata& metadata,
+    const std::unordered_set<UUID, boost::hash<UUID>>& alive_clients) {
+    // Remove those with invalid allocators (memory replicas on unmounted
+    // segments) and local_disk replicas whose owner client is no longer alive.
+    metadata.EraseReplicas([&alive_clients](const Replica& replica) {
+        return replica.has_invalid_mem_handle() ||
+               replica.has_stale_local_disk_client(alive_clients);
     });
 
     // Return true if no valid replicas remain after cleanup
@@ -3469,6 +3511,81 @@ void MasterService::BatchEvict(double evict_ratio_target,
         });
     };
 
+    // --- Offload-on-evict support ---
+    long offload_queued_this_cycle = 0;
+    long offload_deferred_count = 0;
+    long offload_cap_forced_count = 0;    // #keys force-evicted due to cap
+    long offload_push_failed_forced = 0;  // #keys force-evicted on push fail
+    const long offload_cap =
+        offload_on_evict_
+            ? static_cast<long>(offloading_queue_limit_ * kOffloadCapRatio)
+            : 0;
+
+    auto has_local_disk_replica = [](const ObjectMetadata& metadata) {
+        return metadata.HasReplica(&Replica::fn_is_local_disk_replica);
+    };
+
+    // Returns freed bytes. Returns 0 if offload-queued and no additional
+    // replicas were evicted (all MEMORY replicas of the key are now pinned).
+    auto try_evict_or_offload =
+        [&, this](const std::string& key, ObjectMetadata& metadata,
+                  MetadataShardAccessorRW& shard) -> uint64_t {
+        if (!offload_on_evict_) {
+            // Original behavior
+            return metadata.size * evict_replicas(metadata);
+        }
+
+        // LOCAL_DISK replica already exists — safe to delete MEMORY immediately
+        if (has_local_disk_replica(metadata)) {
+            return metadata.size * evict_replicas(metadata);
+        }
+
+        // Force-evict cap: if force_evict enabled and cap reached, force
+        // delete. Warning is aggregated at the end of the cycle to avoid log
+        // flooding.
+        if (offload_force_evict_ && offload_queued_this_cycle >= offload_cap) {
+            offload_cap_forced_count++;
+            return metadata.size * evict_replicas(metadata);
+        }
+
+        // Queue one MEMORY replica for offload; others will be evicted below.
+        bool queued = false;
+        metadata.VisitReplicas(
+            [](const Replica& r) {
+                return r.is_memory_replica() && r.is_completed() &&
+                       r.get_refcnt() == 0;
+            },
+            [this, &key, &shard, &queued, &now](Replica& replica) {
+                if (queued) return;  // only need to pin one replica for offload
+                auto result = PushOffloadingQueue(key, replica);
+                if (result) {
+                    replica.inc_refcnt();
+                    shard->offloading_tasks.emplace(
+                        key, OffloadingTask{replica.id(), now});
+                    queued = true;
+                }
+            });
+
+        if (queued) {
+            offload_queued_this_cycle++;
+            offload_deferred_count++;
+            // Any remaining MEMORY replicas with refcnt==0 are redundant copies
+            // (data survives via the pinned replica → disk). Evict them now to
+            // reclaim memory immediately rather than waiting another cycle.
+            return metadata.size * evict_replicas(metadata);
+        }
+
+        // PushOffloadingQueue failed. Default (data-preserving) behavior is to
+        // skip this cycle — the outer eviction loop will retry after the
+        // offload queue drains. Only force-evict when explicitly opted in, to
+        // prevent silent data loss when the queue is unavailable.
+        if (offload_force_evict_) {
+            offload_push_failed_forced++;
+            return metadata.size * evict_replicas(metadata);
+        }
+        return 0;
+    };
+
     // Randomly select a starting shard to avoid imbalance eviction between
     // shards. No need to use expensive random_device here.
     size_t start_idx = rand() % kNumShards;
@@ -3541,16 +3658,18 @@ void MasterService::BatchEvict(double evict_ratio_target,
                     continue;
                 }
                 if (it->second.lease_timeout <= target_timeout) {
-                    // Evict this object
-                    total_freed_size +=
-                        it->second.size *
-                        evict_replicas(it->second);  // Erase memory replicas
+                    // Evict this object (or defer for offload)
+                    uint64_t freed =
+                        try_evict_or_offload(it->first, it->second, shard);
+                    total_freed_size += freed;
                     if (it->second.IsValid() == false) {
                         it = shard->metadata.erase(it);
                     } else {
                         ++it;
                     }
-                    shard_evicted_count++;
+                    if (freed > 0) {
+                        shard_evicted_count++;
+                    }
                 } else {
                     // second pass candidates
                     no_pin_objects.push_back(it->second.lease_timeout);
@@ -3602,17 +3721,18 @@ void MasterService::BatchEvict(double evict_ratio_target,
                         it->second.lease_timeout <= target_timeout &&
                         !it->second.IsSoftPinned(now) &&
                         can_evict_replicas(it->second)) {
-                        // Evict this object
-                        total_freed_size +=
-                            it->second.size *
-                            evict_replicas(
-                                it->second);  // Erase memory replicas
+                        // Evict this object (or defer for offload)
+                        uint64_t freed =
+                            try_evict_or_offload(it->first, it->second, shard);
+                        total_freed_size += freed;
                         if (it->second.IsValid() == false) {
                             it = shard->metadata.erase(it);
                         } else {
                             ++it;
                         }
-                        evicted_count++;
+                        if (freed > 0) {
+                            evicted_count++;
+                        }
                         target_evict_num--;
                     } else {
                         ++it;
@@ -3652,16 +3772,18 @@ void MasterService::BatchEvict(double evict_ratio_target,
                     // and lease timeout less than or equal to target.
                     if (!it->second.IsSoftPinned(now) ||
                         it->second.lease_timeout <= soft_target_timeout) {
-                        total_freed_size +=
-                            it->second.size *
-                            evict_replicas(
-                                it->second);  // Erase memory replicas
+                        // Evict this object (or defer for offload)
+                        uint64_t freed =
+                            try_evict_or_offload(it->first, it->second, shard);
+                        total_freed_size += freed;
                         if (it->second.IsValid() == false) {
                             it = shard->metadata.erase(it);
                         } else {
                             ++it;
                         }
-                        evicted_count++;
+                        if (freed > 0) {
+                            evicted_count++;
+                        }
                         target_evict_num--;
                     } else {
                         ++it;
@@ -3682,7 +3804,11 @@ void MasterService::BatchEvict(double evict_ratio_target,
         }
     }
 
-    if (evicted_count > 0 || released_discarded_cnt > 0) {
+    if (evicted_count > 0 || released_discarded_cnt > 0 ||
+        offload_deferred_count > 0) {
+        // Offload-deferred counts as partial success: work was done (objects
+        // queued for disk offload), so suppress re-triggering until the next
+        // watermark breach or explicit need_eviction_ signal.
         need_eviction_ = false;
         MasterMetricManager::instance().inc_eviction_success(evicted_count,
                                                              total_freed_size);
@@ -3694,7 +3820,27 @@ void MasterService::BatchEvict(double evict_ratio_target,
         MasterMetricManager::instance().inc_eviction_fail();
     }
     VLOG(1) << "action=evict_objects" << ", evicted_count=" << evicted_count
+            << ", offload_deferred=" << offload_deferred_count
+            << ", offload_cap_forced=" << offload_cap_forced_count
+            << ", offload_push_failed_forced=" << offload_push_failed_forced
             << ", total_freed_size=" << total_freed_size;
+    if (offload_on_evict_ && evicted_count == 0 && offload_deferred_count > 0) {
+        LOG(WARNING) << "[EVICT] No memory freed this cycle; "
+                     << offload_deferred_count
+                     << " objects deferred for disk offload. "
+                        "Consider lowering eviction_high_watermark_ratio.";
+    }
+    if (offload_cap_forced_count > 0) {
+        LOG(WARNING) << "[EVICT] Offload cap (" << offload_cap
+                     << ") reached; force-evicted " << offload_cap_forced_count
+                     << " object(s) without disk offload this cycle.";
+    }
+    if (offload_push_failed_forced > 0) {
+        LOG(WARNING) << "[EVICT] PushOffloadingQueue failed for "
+                     << offload_push_failed_forced
+                     << " object(s); force-evicted without disk offload "
+                        "(offload_force_evict=true).";
+    }
 }
 
 void MasterService::ClientMonitorFunc() {
@@ -3771,9 +3917,15 @@ void MasterService::ClientMonitorFunc() {
             }  // Release the mutex before long-running ClearInvalidHandles and
                // avoid deadlocks
 
-            if (!unmount_segments.empty()) {
-                ClearInvalidHandles();
+            // Always clean up invalid handles when there are expired clients,
+            // even if no memory segments were unmounted. This is necessary
+            // to clean up local_disk replicas whose owner client has expired.
+            ClearInvalidHandles();
 
+            // Commit unmount of memory segments and clean up local_disk
+            // segments for expired clients. Both require the exclusive
+            // segment lock.
+            {
                 ScopedSegmentAccess segment_access =
                     segment_manager_.getSegmentAccess();
                 for (size_t i = 0; i < unmount_segments.size(); i++) {
@@ -3783,6 +3935,9 @@ void MasterService::ClientMonitorFunc() {
                               << ", segment_name=" << segment_names[i]
                               << ", action=unmount_expired_segment";
                 }
+                for (auto& client_id : expired_clients) {
+                    segment_access.UnmountLocalDiskSegment(client_id);
+                }
             }
         }
 

mooncake-store/src/real_client.cpp

@@ -62,6 +62,87 @@ struct PreparedRangedReadRequest {
     bool has_any_valid_fragment = false;
 };
 
+size_t sum_value_sizes(const std::vector<std::span<const char>> &values) {
+    size_t total = 0;
+    for (const auto &value : values) {
+        total += value.size_bytes();
+    }
+    return total;
+}
+
+size_t sum_sizes(const std::vector<size_t> &sizes) {
+    size_t total = 0;
+    for (size_t size : sizes) {
+        total += size;
+    }
+    return total;
+}
+
+size_t sum_successful_sizes(const std::vector<int> &results,
+                            const std::vector<size_t> &sizes) {
+    size_t total = 0;
+    for (size_t i = 0; i < results.size() && i < sizes.size(); ++i) {
+        if (results[i] == 0) {
+            total += sizes[i];
+        }
+    }
+    return total;
+}
+
+size_t sum_successful_nested_sizes(
+    const std::vector<int> &results,
+    const std::vector<std::vector<size_t>> &nested_sizes) {
+    size_t total = 0;
+    for (size_t i = 0; i < results.size() && i < nested_sizes.size(); ++i) {
+        if (results[i] == 0) {
+            total += sum_sizes(nested_sizes[i]);
+        }
+    }
+    return total;
+}
+
+size_t sum_positive_results(const std::vector<int64_t> &results) {
+    size_t total = 0;
+    for (int64_t result : results) {
+        if (result > 0) {
+            total += static_cast<size_t>(result);
+        }
+    }
+    return total;
+}
+
+size_t sum_positive_results(const std::vector<int> &results) {
+    size_t total = 0;
+    for (int result : results) {
+        if (result > 0) {
+            total += static_cast<size_t>(result);
+        }
+    }
+    return total;
+}
+
+size_t sum_positive_ranges(
+    const std::vector<std::vector<std::vector<int64_t>>> &results) {
+    size_t total = 0;
+    for (const auto &key_rows : results) {
+        for (const auto &row : key_rows) {
+            total += sum_positive_results(row);
+        }
+    }
+    return total;
+}
+
+size_t sum_buffer_handle_sizes(
+    const std::vector<std::shared_ptr<BufferHandle>> &buffers) {
+    size_t total = 0;
+    for (const auto &buffer : buffers) {
+        if (buffer != nullptr) {
+            total += buffer->size();
+        }
+    }
+    return total;
+}
+
 PreparedRangedReadRequest prepare_ranged_read_request(
     size_t buffer_count, const std::vector<std::vector<std::string>> &all_keys,
     const std::vector<std::vector<std::vector<size_t>>> &all_dst_offsets,
@@ -494,7 +575,7 @@ tl::expected<void, ErrorCode> RealClient::setup_internal(
             hostname.substr(0, colon_pos + 1) + std::to_string(local_rpc_port);
         auto client_opt = mooncake::Client::Create(
             this->local_hostname, metadata_server, protocol, device_name,
-            master_server_addr, transfer_engine);
+            master_server_addr, transfer_engine, {{"client_mode", "real"}});
         if (!client_opt) {
             LOG(ERROR) << "Failed to create client";
             return tl::unexpected(ErrorCode::INVALID_PARAMS);
@@ -523,7 +604,7 @@ tl::expected<void, ErrorCode> RealClient::setup_internal(
                 hostname + ":" + std::to_string(local_rpc_port);
             auto client_opt = mooncake::Client::Create(
                 this->local_hostname, metadata_server, protocol, device_name,
-                master_server_addr, transfer_engine);
+                master_server_addr, transfer_engine, {{"client_mode", "real"}});
             if (client_opt) {
                 client_ = *client_opt;
                 success = true;
@@ -1084,8 +1165,17 @@ tl::expected<void, ErrorCode> RealClient::put_dummy_helper(
 
 int RealClient::put(const std::string &key, std::span<const char> value,
                     const ReplicateConfig &config) {
-    return to_py_ret(
-        put_internal(key, value, config, client_buffer_allocator_));
+    auto result = execute_timed_operation<tl::expected<void, ErrorCode>>(
+        [&]() {
+            return put_internal(key, value, config, client_buffer_allocator_);
+        },
+        [](const auto &ret) { return ret.has_value(); },
+        [&](uint64_t latency_us, const auto &) {
+            client_->ObserveTransferOperation(TransferOperationKind::kWrite,
+                                              "put", value.size_bytes(),
+                                              latency_us);
+        });
+    return to_py_ret(result);
 }
 
 tl::expected<void, ErrorCode> RealClient::put_batch_internal(
@@ -1174,8 +1264,18 @@ tl::expected<void, ErrorCode> RealClient::put_batch_dummy_helper(
 int RealClient::put_batch(const std::vector<std::string> &keys,
                           const std::vector<std::span<const char>> &values,
                           const ReplicateConfig &config) {
-    return to_py_ret(
-        put_batch_internal(keys, values, config, client_buffer_allocator_));
+    auto result = execute_timed_operation<tl::expected<void, ErrorCode>>(
+        [&]() {
+            return put_batch_internal(keys, values, config,
+                                      client_buffer_allocator_);
+        },
+        [](const auto &ret) { return ret.has_value(); },
+        [&](uint64_t latency_us, const auto &) {
+            client_->ObserveTransferOperation(
+                TransferOperationKind::kWrite, "put_batch",
+                sum_value_sizes(values), latency_us);
+        });
+    return to_py_ret(result);
 }
 
 tl::expected<void, ErrorCode> RealClient::put_parts_internal(
@@ -1256,8 +1356,18 @@ tl::expected<void, ErrorCode> RealClient::put_parts_dummy_helper(
 int RealClient::put_parts(const std::string &key,
                           std::vector<std::span<const char>> values,
                           const ReplicateConfig &config) {
-    return to_py_ret(
-        put_parts_internal(key, values, config, client_buffer_allocator_));
+    auto result = execute_timed_operation<tl::expected<void, ErrorCode>>(
+        [&]() {
+            return put_parts_internal(key, values, config,
+                                      client_buffer_allocator_);
+        },
+        [](const auto &ret) { return ret.has_value(); },
+        [&](uint64_t latency_us, const auto &) {
+            client_->ObserveTransferOperation(
+                TransferOperationKind::kWrite, "put_parts",
+                sum_value_sizes(values), latency_us);
+        });
+    return to_py_ret(result);
 }
 
 tl::expected<void, ErrorCode> RealClient::remove_internal(
@@ -1854,7 +1964,14 @@ std::shared_ptr<BufferHandle> RealClient::get_buffer_internal(
 
 // Implementation of get_buffer method
 std::shared_ptr<BufferHandle> RealClient::get_buffer(const std::string &key) {
-    return get_buffer_internal(key, client_buffer_allocator_);
+    return execute_timed_operation<std::shared_ptr<BufferHandle>>(
+        [&]() { return get_buffer_internal(key, client_buffer_allocator_); },
+        [](const auto &buffer) { return buffer != nullptr; },
+        [&](uint64_t latency_us, const auto &buffer) {
+            client_->ObserveTransferOperation(TransferOperationKind::kRead,
+                                              "get_buffer", buffer->size(),
+                                              latency_us);
+        });
 }
 
 tl::expected<std::tuple<uint64_t, size_t>, ErrorCode>
@@ -2144,7 +2261,14 @@ RealClient::batch_get_buffer_internal(
 // Implementation of batch_get_buffer method
 std::vector<std::shared_ptr<BufferHandle>> RealClient::batch_get_buffer(
     const std::vector<std::string> &keys) {
-    return batch_get_buffer_internal(keys);
+    return execute_timed_operation<std::vector<std::shared_ptr<BufferHandle>>>(
+        [&]() { return batch_get_buffer_internal(keys); },
+        [](const auto &) { return true; },
+        [&](uint64_t latency_us, const auto &buffers) {
+            client_->ObserveTransferOperation(
+                TransferOperationKind::kRead, "batch_get_buffer",
+                sum_buffer_handle_sizes(buffers), latency_us);
+        });
 }
 
 tl::expected<void, ErrorCode> RealClient::register_buffer_internal(
@@ -2299,7 +2423,17 @@ tl::expected<int64_t, ErrorCode> RealClient::get_into_range_internal(
 
 int64_t RealClient::get_into(const std::string &key, void *buffer,
                              size_t size) {
-    return to_py_ret(get_into_range_internal(key, buffer, 0, 0, size, true));
+    auto result = execute_timed_operation<tl::expected<int64_t, ErrorCode>>(
+        [&]() {
+            return get_into_range_internal(key, buffer, 0, 0, size, true);
+        },
+        [](const auto &ret) { return ret.has_value(); },
+        [&](uint64_t latency_us, const auto &ret) {
+            client_->ObserveTransferOperation(
+                TransferOperationKind::kRead, "get_into",
+                static_cast<uint64_t>(ret.value()), latency_us);
+        });
+    return to_py_ret(result);
 }
 
 std::vector<std::vector<std::vector<tl::expected<int64_t, ErrorCode>>>>
@@ -2419,8 +2553,20 @@ std::vector<std::vector<std::vector<int64_t>>> RealClient::get_into_ranges(
     const std::vector<std::vector<std::vector<size_t>>> &all_dst_offsets,
     const std::vector<std::vector<std::vector<size_t>>> &all_src_offsets,
     const std::vector<std::vector<std::vector<size_t>>> &all_sizes) {
-    return convert_ranged_read_results(get_into_ranges_internal(
-        buffers, all_keys, all_dst_offsets, all_src_offsets, all_sizes));
+    auto results =
+        execute_timed_operation<std::vector<std::vector<std::vector<int64_t>>>>(
+            [&]() {
+                return convert_ranged_read_results(
+                    get_into_ranges_internal(buffers, all_keys, all_dst_offsets,
+                                             all_src_offsets, all_sizes));
+            },
+            [](const auto &) { return true; },
+            [&](uint64_t latency_us, const auto &ret) {
+                client_->ObserveTransferOperation(
+                    TransferOperationKind::kRead, "get_into_ranges",
+                    sum_positive_ranges(ret), latency_us);
+            });
+    return results;
 }
 
 std::string RealClient::get_hostname() const { return local_hostname; }
@@ -2429,7 +2575,21 @@ std::vector<int> RealClient::batch_put_from(
     const std::vector<std::string> &keys, const std::vector<void *> &buffers,
     const std::vector<size_t> &sizes, const ReplicateConfig &config) {
     auto internal_results =
-        batch_put_from_internal(keys, buffers, sizes, config);
+        execute_timed_operation<std::vector<tl::expected<void, ErrorCode>>>(
+            [&]() {
+                return batch_put_from_internal(keys, buffers, sizes, config);
+            },
+            [](const auto &) { return true; },
+            [&](uint64_t latency_us, const auto &ret) {
+                std::vector<int> py_results;
+                py_results.reserve(ret.size());
+                for (const auto &item : ret) {
+                    py_results.push_back(to_py_ret(item));
+                }
+                client_->ObserveTransferOperation(
+                    TransferOperationKind::kWrite, "batch_put_from",
+                    sum_successful_sizes(py_results, sizes), latency_us);
+            });
     std::vector<int> results;
     results.reserve(internal_results.size());
 
@@ -2572,7 +2732,14 @@ tl::expected<void, ErrorCode> RealClient::put_from_internal(
 
 int RealClient::put_from(const std::string &key, void *buffer, size_t size,
                          const ReplicateConfig &config) {
-    return to_py_ret(put_from_internal(key, buffer, size, config));
+    auto result = execute_timed_operation<tl::expected<void, ErrorCode>>(
+        [&]() { return put_from_internal(key, buffer, size, config); },
+        [](const auto &ret) { return ret.has_value(); },
+        [&](uint64_t latency_us, const auto &) {
+            client_->ObserveTransferOperation(TransferOperationKind::kWrite,
+                                              "put_from", size, latency_us);
+        });
+    return to_py_ret(result);
 }
 
 // --- Upsert implementations ---
@@ -2613,8 +2780,18 @@ tl::expected<void, ErrorCode> RealClient::upsert_internal(
 
 int RealClient::upsert(const std::string &key, std::span<const char> value,
                        const ReplicateConfig &config) {
-    return to_py_ret(
-        upsert_internal(key, value, config, client_buffer_allocator_));
+    auto result = execute_timed_operation<tl::expected<void, ErrorCode>>(
+        [&]() {
+            return upsert_internal(key, value, config,
+                                   client_buffer_allocator_);
+        },
+        [](const auto &ret) { return ret.has_value(); },
+        [&](uint64_t latency_us, const auto &) {
+            client_->ObserveTransferOperation(TransferOperationKind::kWrite,
+                                              "upsert", value.size_bytes(),
+                                              latency_us);
+        });
+    return to_py_ret(result);
 }
 
 tl::expected<void, ErrorCode> RealClient::upsert_dummy_helper(
@@ -2664,7 +2841,14 @@ tl::expected<void, ErrorCode> RealClient::upsert_from_internal(
 
 int RealClient::upsert_from(const std::string &key, void *buffer, size_t size,
                             const ReplicateConfig &config) {
-    return to_py_ret(upsert_from_internal(key, buffer, size, config));
+    auto result = execute_timed_operation<tl::expected<void, ErrorCode>>(
+        [&]() { return upsert_from_internal(key, buffer, size, config); },
+        [](const auto &ret) { return ret.has_value(); },
+        [&](uint64_t latency_us, const auto &) {
+            client_->ObserveTransferOperation(TransferOperationKind::kWrite,
+                                              "upsert_from", size, latency_us);
+        });
+    return to_py_ret(result);
 }
 
 std::vector<tl::expected<void, ErrorCode>>
@@ -2710,7 +2894,21 @@ std::vector<int> RealClient::batch_upsert_from(
     const std::vector<std::string> &keys, const std::vector<void *> &buffers,
     const std::vector<size_t> &sizes, const ReplicateConfig &config) {
     auto internal_results =
-        batch_upsert_from_internal(keys, buffers, sizes, config);
+        execute_timed_operation<std::vector<tl::expected<void, ErrorCode>>>(
+            [&]() {
+                return batch_upsert_from_internal(keys, buffers, sizes, config);
+            },
+            [](const auto &) { return true; },
+            [&](uint64_t latency_us, const auto &ret) {
+                std::vector<int> py_results;
+                py_results.reserve(ret.size());
+                for (const auto &item : ret) {
+                    py_results.push_back(to_py_ret(item));
+                }
+                client_->ObserveTransferOperation(
+                    TransferOperationKind::kWrite, "batch_upsert_from",
+                    sum_successful_sizes(py_results, sizes), latency_us);
+            });
     std::vector<int> results;
     results.reserve(internal_results.size());
     for (const auto &result : internal_results) {
@@ -2834,8 +3032,18 @@ tl::expected<void, ErrorCode> RealClient::upsert_parts_internal(
 int RealClient::upsert_parts(const std::string &key,
                              std::vector<std::span<const char>> values,
                              const ReplicateConfig &config) {
-    return to_py_ret(
-        upsert_parts_internal(key, values, config, client_buffer_allocator_));
+    auto result = execute_timed_operation<tl::expected<void, ErrorCode>>(
+        [&]() {
+            return upsert_parts_internal(key, values, config,
+                                         client_buffer_allocator_);
+        },
+        [](const auto &ret) { return ret.has_value(); },
+        [&](uint64_t latency_us, const auto &) {
+            client_->ObserveTransferOperation(
+                TransferOperationKind::kWrite, "upsert_parts",
+                sum_value_sizes(values), latency_us);
+        });
+    return to_py_ret(result);
 }
 
 tl::expected<void, ErrorCode> RealClient::upsert_parts_dummy_helper(
@@ -2938,8 +3146,18 @@ tl::expected<void, ErrorCode> RealClient::upsert_batch_dummy_helper(
 int RealClient::upsert_batch(const std::vector<std::string> &keys,
                              const std::vector<std::span<const char>> &values,
                              const ReplicateConfig &config) {
-    return to_py_ret(
-        upsert_batch_internal(keys, values, config, client_buffer_allocator_));
+    auto result = execute_timed_operation<tl::expected<void, ErrorCode>>(
+        [&]() {
+            return upsert_batch_internal(keys, values, config,
+                                         client_buffer_allocator_);
+        },
+        [](const auto &ret) { return ret.has_value(); },
+        [&](uint64_t latency_us, const auto &) {
+            client_->ObserveTransferOperation(
+                TransferOperationKind::kWrite, "upsert_batch",
+                sum_value_sizes(values), latency_us);
+        });
+    return to_py_ret(result);
 }
 
 // --- End Upsert implementations ---
@@ -2947,7 +3165,20 @@ int RealClient::upsert_batch(const std::vector<std::string> &keys,
 std::vector<int64_t> RealClient::batch_get_into(
     const std::vector<std::string> &keys, const std::vector<void *> &buffers,
     const std::vector<size_t> &sizes) {
-    auto internal_results = batch_get_into_internal(keys, buffers, sizes);
+    auto internal_results =
+        execute_timed_operation<std::vector<tl::expected<int64_t, ErrorCode>>>(
+            [&]() { return batch_get_into_internal(keys, buffers, sizes); },
+            [](const auto &) { return true; },
+            [&](uint64_t latency_us, const auto &ret) {
+                std::vector<int64_t> py_results;
+                py_results.reserve(ret.size());
+                for (const auto &item : ret) {
+                    py_results.push_back(to_py_ret(item));
+                }
+                client_->ObserveTransferOperation(
+                    TransferOperationKind::kRead, "batch_get_into",
+                    sum_positive_results(py_results), latency_us);
+            });
     std::vector<int64_t> results;
     results.reserve(internal_results.size());
 
@@ -3351,6 +3582,7 @@ int RealClient::put_from_with_metadata(const std::string &key, void *buffer,
                                        void *metadata_buffer, size_t size,
                                        size_t metadata_size,
                                        const ReplicateConfig &config) {
+    const auto start_time = std::chrono::steady_clock::now();
     // NOTE: The buffer address must be previously registered with
     // register_buffer() for zero-copy RDMA operations to work correctly
     if (config.prefer_alloc_in_same_node) {
@@ -3393,6 +3625,10 @@ int RealClient::put_from_with_metadata(const std::string &key, void *buffer,
                    << toString(put_result.error());
         return -toInt(put_result.error());
     }
+
+    client_->ObserveTransferOperation(
+        TransferOperationKind::kWrite, "put_from_with_metadata",
+        size + metadata_size, elapsed_us_since(start_time));
     return 0;
 }
 
@@ -3401,10 +3637,24 @@ std::vector<int> RealClient::batch_put_from_multi_buffers(
     const std::vector<std::vector<void *>> &all_buffers,
     const std::vector<std::vector<size_t>> &sizes,
     const ReplicateConfig &config) {
-    auto start = std::chrono::steady_clock::now();
-
     auto internal_results =
-        batch_put_from_multi_buffers_internal(keys, all_buffers, sizes, config);
+        execute_timed_operation<std::vector<tl::expected<void, ErrorCode>>>(
+            [&]() {
+                return batch_put_from_multi_buffers_internal(keys, all_buffers,
+                                                             sizes, config);
+            },
+            [](const auto &) { return true; },
+            [&](uint64_t latency_us, const auto &ret) {
+                std::vector<int> py_results;
+                py_results.reserve(ret.size());
+                for (const auto &item : ret) {
+                    py_results.push_back(to_py_ret(item));
+                }
+                client_->ObserveTransferOperation(
+                    TransferOperationKind::kWrite,
+                    "batch_put_from_multi_buffers",
+                    sum_successful_nested_sizes(py_results, sizes), latency_us);
+            });
     std::vector<int> results;
     results.reserve(internal_results.size());
 
@@ -3412,10 +3662,6 @@ std::vector<int> RealClient::batch_put_from_multi_buffers(
         results.push_back(to_py_ret(result));
     }
 
-    auto duration_call = std::chrono::duration_cast<std::chrono::microseconds>(
-        std::chrono::steady_clock::now() - start);
-    VLOG(1) << "batch_put_from_multi_buffers: " << duration_call.count()
-            << " us";
     return results;
 }
 
@@ -3461,19 +3707,30 @@ std::vector<int> RealClient::batch_get_into_multi_buffers(
     const std::vector<std::vector<void *>> &all_buffers,
     const std::vector<std::vector<size_t>> &all_sizes,
     bool prefer_alloc_in_same_node) {
-    auto start = std::chrono::steady_clock::now();
-    auto internal_results = batch_get_into_multi_buffers_internal(
-        keys, all_buffers, all_sizes, prefer_alloc_in_same_node);
+    auto internal_results =
+        execute_timed_operation<std::vector<tl::expected<int64_t, ErrorCode>>>(
+            [&]() {
+                return batch_get_into_multi_buffers_internal(
+                    keys, all_buffers, all_sizes, prefer_alloc_in_same_node);
+            },
+            [](const auto &) { return true; },
+            [&](uint64_t latency_us, const auto &ret) {
+                std::vector<int> py_results;
+                py_results.reserve(ret.size());
+                for (const auto &item : ret) {
+                    py_results.push_back(to_py_ret(item));
+                }
+                client_->ObserveTransferOperation(
+                    TransferOperationKind::kRead,
+                    "batch_get_into_multi_buffers",
+                    sum_positive_results(py_results), latency_us);
+            });
     std::vector<int> results;
     results.reserve(internal_results.size());
 
     for (const auto &result : internal_results) {
         results.push_back(to_py_ret(result));
     }
-    auto duration_call = std::chrono::duration_cast<std::chrono::microseconds>(
-        std::chrono::steady_clock::now() - start);
-    VLOG(1) << "batch_get_into_multi_buffers: " << duration_call.count()
-            << " us";
     return results;
 }
 
@@ -4009,6 +4266,18 @@ ClientRequester::ClientRequester() {
         pool_conf.client_config.socket_config =
             coro_io::ib_socket_t::config_t{};
     }
+    // Configure reasonable retry limits for SSD offload RPC connections.
+    // - connect_retry_count: Maximum connection retry attempts (default: 3)
+    // - reconnect_wait_time: Wait time between retries (default: 1000ms)
+    // - host_alive_detect_duration: Duration for background alive detection.
+    //   Set to 0 to disable infinite background reconnection attempts when
+    //   a Store node goes down. This prevents continuous "Connection refused"
+    //   logs. When Master cleans up stale local_disk replicas (via
+    //   CleanupStaleHandles), new requests won't route to dead nodes anyway.
+    pool_conf.connect_retry_count = 3;
+    pool_conf.reconnect_wait_time = std::chrono::milliseconds{1000};
+    pool_conf.host_alive_detect_duration = std::chrono::milliseconds{0};
+
     client_pools_ =
         std::make_shared<coro_io::client_pools<coro_rpc::coro_rpc_client>>(
             pool_conf);

mooncake-store/src/real_client_main.cpp

@@ -91,6 +91,10 @@ int main(int argc, char *argv[]) {
     mooncake::ResourceTracker::getInstance();
 
     gflags::ParseCommandLineFlags(&argc, &argv, true);
+    if (!FLAGS_log_dir.empty()) {
+        google::InitGoogleLogging(argv[0]);
+    }
+
     size_t global_segment_size = string_to_byte_size(FLAGS_global_segment_size);
 #ifdef USE_ASCEND_DIRECT
     // just set to true, does not affect GPU process.

mooncake-store/src/segment.cpp

@@ -273,6 +273,15 @@ ErrorCode ScopedSegmentAccess::GetClientSegments(
     return ErrorCode::OK;
 }
 
+void ScopedSegmentAccess::UnmountLocalDiskSegment(const UUID& client_id) {
+    auto it = segment_manager_->client_local_disk_segment_.find(client_id);
+    if (it != segment_manager_->client_local_disk_segment_.end()) {
+        segment_manager_->client_local_disk_segment_.erase(it);
+        LOG(INFO) << "client_id=" << client_id
+                  << ", action=unmount_local_disk_segment";
+    }
+}
+
 ErrorCode ScopedSegmentAccess::GetAllSegments(
     std::vector<std::string>& all_segments) {
     all_segments.clear();

mooncake-store/src/storage_backend.cpp

@@ -49,8 +49,9 @@ FilePerKeyConfig FilePerKeyConfig::FromEnvironment() {
 
     config.fsdir = GetEnvStringOr("MOONCAKE_OFFLOAD_FSDIR", config.fsdir);
 
-    config.enable_eviction =
-        GetEnvOr<bool>("ENABLE_EVICTION", config.enable_eviction);
+    config.enable_eviction = GetEnvOr<bool>(
+        "MOONCAKE_OFFLOAD_ENABLE_EVICTION",
+        GetEnvOr<bool>("ENABLE_EVICTION", config.enable_eviction));
 
     return config;
 }
@@ -64,11 +65,14 @@ BucketBackendConfig BucketBackendConfig::FromEnvironment() {
     config.bucket_size_limit = GetEnvOr<int64_t>(
         "MOONCAKE_OFFLOAD_BUCKET_SIZE_LIMIT_BYTES", config.bucket_size_limit);
 
-    config.max_total_size = GetEnvOr<int64_t>("MOONCAKE_BUCKET_MAX_TOTAL_SIZE",
-                                              config.max_total_size);
+    config.max_total_size =
+        GetEnvOr<int64_t>("MOONCAKE_OFFLOAD_BUCKET_MAX_TOTAL_SIZE",
+                          GetEnvOr<int64_t>("MOONCAKE_BUCKET_MAX_TOTAL_SIZE",
+                                            config.max_total_size));
 
-    const auto policy_str =
-        GetEnvStringOr("MOONCAKE_BUCKET_EVICTION_POLICY", "none");
+    const auto policy_str = GetEnvStringOr(
+        "MOONCAKE_OFFLOAD_BUCKET_EVICTION_POLICY",
+        GetEnvStringOr("MOONCAKE_BUCKET_EVICTION_POLICY", "none"));
     if (policy_str == "fifo") {
         config.eviction_policy = BucketEvictionPolicy::FIFO;
     } else if (policy_str == "lru") {
@@ -1718,6 +1722,18 @@ tl::expected<void, ErrorCode> BucketStorageBackend::Init() {
                       << orphaned_space_freed << " bytes";
         }
 
+        // When max_total_size is not explicitly set (<= 0), default to 90% of
+        // the physical disk capacity to match FilePerKey backend behavior.
+        if (bucket_backend_config_.max_total_size <= 0) {
+            constexpr double kDefaultQuotaPercentage = 0.9;
+            const auto space_info = fs::space(storage_path_);
+            bucket_backend_config_.max_total_size = static_cast<int64_t>(
+                space_info.capacity * kDefaultQuotaPercentage);
+            LOG(INFO) << "Bucket backend max_total_size not set; using "
+                      << kDefaultQuotaPercentage * 100 << "% of disk capacity: "
+                      << bucket_backend_config_.max_total_size << " bytes";
+        }
+
         bucket_id_generator_.emplace(max_bucket_id);
         if (max_bucket_id == BucketIdGenerator::INIT_NEW_START_ID) {
             LOG(INFO) << "Initialized BucketIdGenerator with fresh start. "
@@ -2187,8 +2203,7 @@ BucketStorageBackend::PendingEviction BucketStorageBackend::PrepareEviction(
     int64_t required_size) {
     PendingEviction result;
 
-    if (bucket_backend_config_.eviction_policy == BucketEvictionPolicy::NONE ||
-        bucket_backend_config_.max_total_size <= 0) {
+    if (bucket_backend_config_.eviction_policy == BucketEvictionPolicy::NONE) {
         return result;
     }
 

mooncake-store/src/transfer_task.cpp

@@ -4,6 +4,8 @@
 
 #include <algorithm>
 #include <cstdlib>
+#include <sstream>
+#include <vector>
 #include "transfer_engine.h"
 #include "transport/transport.h"
 
@@ -223,9 +225,13 @@ bool TransferEngineOperationState::is_completed() {
 }
 
 void TransferEngineOperationState::check_task_status() {
-    // Check all transfers in the batch
-    bool all_completed = true;
-    bool has_failure = false;
+    // Check all transfers in the batch.
+    // Wait for ALL tasks to reach a terminal state before setting the result,
+    // even if some have already failed. This prevents the caller from seeing
+    // "completed" while background transfers are still in progress, which
+    // could cause issues when freeBatchID is called in the destructor.
+    bool all_terminated = true;
+    std::vector<size_t> failed_task_ids;
 
     for (size_t i = 0; i < batch_size_; ++i) {
         TransferStatus status;
@@ -240,38 +246,45 @@ void TransferEngineOperationState::check_task_status() {
 
         switch (status.s) {
             case TransferStatusEnum::COMPLETED:
-                // This transfer is done, continue checking others
+                // This transfer is done successfully
                 break;
             case TransferStatusEnum::FAILED:
             case TransferStatusEnum::CANCELED:
             case TransferStatusEnum::INVALID:
 #ifndef USE_ASCEND_DIRECT
-                LOG(ERROR) << "Transfer failed for batch " << batch_id_
-                           << " task " << i << " with status "
-                           << static_cast<int>(status.s);
+                VLOG(1) << "Transfer failed for batch " << batch_id_ << " task "
+                        << i << " with status " << static_cast<int>(status.s);
 #endif
-                has_failure = true;
+                failed_task_ids.push_back(i);
                 break;
             default:
-                // Transfer is still pending (PENDING, RUNNING, etc.)
-                all_completed = false;
+                // Transfer is still in progress (WAITING, PENDING, etc.)
+                all_terminated = false;
                 break;
         }
     }
 
-    if (has_failure) {
-        VLOG(1) << "Setting batch " << batch_id_
-                << " result to TRANSFER_FAIL due to task failures";
-        set_result_internal(ErrorCode::TRANSFER_FAIL);
+    if (!all_terminated) {
+        // Some tasks are still in progress; wait for next poll iteration.
+        // Do NOT set result yet, even if some tasks have already failed.
         return;
     }
 
-    if (all_completed) {
-        set_result_internal(ErrorCode::OK);
-        return;
+    // All tasks have reached a terminal state.
+    ErrorCode ec = ErrorCode::OK;
+    if (!failed_task_ids.empty()) {
+        std::ostringstream oss;
+        for (size_t j = 0; j < failed_task_ids.size(); ++j) {
+            if (j > 0) oss << ", ";
+            oss << failed_task_ids[j];
+        }
+        LOG(ERROR) << "Batch " << batch_id_
+                   << " completed with task failures: task_ids=[" << oss.str()
+                   << "]";
+        ec = ErrorCode::TRANSFER_FAIL;
     }
 
-    return;
+    set_result_internal(ec);
 }
 
 void TransferEngineOperationState::set_result_internal(ErrorCode error_code) {
@@ -416,10 +429,17 @@ TransferSubmitter::TransferSubmitter(TransferEngine& engine,
       memcpy_pool_(std::make_unique<MemcpyWorkerPool>()),
       fileread_pool_(std::make_unique<FilereadWorkerPool>(backend)),
       transfer_metric_(transfer_metric) {
-    // Read MC_STORE_MEMCPY environment variable, default to false (disabled)
+    // Read MC_STORE_MEMCPY environment variable.
+    // When not set, auto-detect based on transport type:
+    //   - TCP-only environment: enable memcpy (avoids TCP loopback overhead)
+    //   - RDMA/other transports: disable memcpy (RDMA is more efficient)
     const char* env_value = std::getenv("MC_STORE_MEMCPY");
     if (env_value == nullptr) {
-        memcpy_enabled_ = false;  // Default: disabled
+        memcpy_enabled_ = engine_.isTcpOnly();
+        LOG(INFO) << "MC_STORE_MEMCPY not set, auto-detected: "
+                  << (memcpy_enabled_ ? "TCP-only environment, memcpy enabled"
+                                      : "non-TCP transport available, memcpy "
+                                        "disabled");
     } else {
         std::string env_str(env_value);
         // Convert to lowercase for case-insensitive comparison

mooncake-store/tests/CMakeLists.txt

@@ -39,6 +39,7 @@ add_store_test(eviction_strategy_test eviction_strategy_test.cpp)
 add_store_test(master_service_test master_service_test.cpp)
 add_store_test(batch_remove_test batch_remove_test.cpp)
 add_store_test(master_service_ssd_test master_service_ssd_test.cpp)
+add_store_test(offload_on_evict_test offload_on_evict_test.cpp)
 add_store_test(master_service_ssd_test_for_snapshot
                ha/snapshot/master_service_ssd_test_for_snapshot.cpp)
 add_store_test(client_integration_test client_integration_test.cpp)

mooncake-store/tests/client_metrics_test.cpp

@@ -45,6 +45,8 @@ TEST_F(ClientMetricsTest, TransferMetricsSummaryTest) {
     // Check byte formatting
     EXPECT_TRUE(summary.find("Total Read: 1.00 KB") != std::string::npos);
     EXPECT_TRUE(summary.find("Total Write: 2.00 MB") != std::string::npos);
+    EXPECT_TRUE(summary.find("Average Read Throughput:") != std::string::npos);
+    EXPECT_TRUE(summary.find("Average Write Throughput:") != std::string::npos);
 
     // Check latency summaries
     EXPECT_TRUE(summary.find("Get: count=3") != std::string::npos);
@@ -109,15 +111,23 @@ TEST_F(ClientMetricsTest, ClientMetricsSummaryTest) {
     std::array<std::string, 1> exist_key_label = {"ExistKey"};
     metrics.master_client_metric.rpc_count.inc(exist_key_label);
     metrics.master_client_metric.rpc_latency.observe(exist_key_label, 180);
+    metrics.ObserveTransferOperation(TransferOperationKind::kRead, "get_buffer",
+                                     2 * 1024, 220);
+    metrics.ObserveTransferOperation(TransferOperationKind::kWrite, "put_batch",
+                                     4 * 1024, 420);
 
     std::string summary = metrics.summary_metrics();
 
-    // Should contain both transfer and RPC metrics
+    // Should contain transfer, RPC, and interface metrics
     EXPECT_TRUE(summary.find("Transfer Metrics Summary") != std::string::npos);
     EXPECT_TRUE(summary.find("RPC Metrics Summary") != std::string::npos);
+    EXPECT_TRUE(summary.find("Interface Operation Metrics Summary") !=
+                std::string::npos);
     EXPECT_TRUE(summary.find("Total Read: 5.00 MB") != std::string::npos);
     EXPECT_TRUE(summary.find("Total Write: 10.00 MB") != std::string::npos);
     EXPECT_TRUE(summary.find("ExistKey: count=1") != std::string::npos);
+    EXPECT_TRUE(summary.find("get_buffer: count=1") != std::string::npos);
+    EXPECT_TRUE(summary.find("put_batch: count=1") != std::string::npos);
 
     std::cout << "Full Client Metrics Summary:\n" << summary << std::endl;
 }
@@ -175,6 +185,33 @@ TEST_F(ClientMetricsTest, CompareWithSerializedMetrics) {
                 summary.find("No data") != std::string::npos);
 }
 
+TEST_F(ClientMetricsTest, BandwidthSummaryRespectsEnvFlag) {
+    setenv("MC_STORE_CLIENT_METRIC_BANDWIDTH", "0", 1);
+    auto metrics = ClientMetric::Create();
+    ASSERT_NE(metrics, nullptr);
+
+    metrics->transfer_metric.total_read_bytes.inc(1024);
+    std::string summary = metrics->summary_metrics();
+    EXPECT_TRUE(summary.find("Average Read Throughput:") == std::string::npos);
+
+    unsetenv("MC_STORE_CLIENT_METRIC_BANDWIDTH");
+}
+
+TEST_F(ClientMetricsTest, SummaryCanOmitMasterRpcMetrics) {
+    auto metrics = ClientMetric::Create({}, false);
+    ASSERT_NE(metrics, nullptr);
+
+    metrics->ObserveTransferOperation(TransferOperationKind::kRead,
+                                      "get_buffer", 1024, 200);
+    std::string summary = metrics->summary_metrics();
+    std::string serialized;
+    metrics->serialize(serialized);
+
+    EXPECT_TRUE(summary.find("RPC Metrics Summary") == std::string::npos);
+    EXPECT_TRUE(serialized.find("mooncake_client_rpc_count") ==
+                std::string::npos);
+}
+
 TEST_F(ClientMetricsTest, SerializeWithDynamicLabels) {
     auto verify = [](const std::string& str) {
         EXPECT_TRUE(str.find("instance_id=\"12345\"") != std::string::npos);

mooncake-store/tests/file_storage_test.cpp

@@ -26,7 +26,8 @@ class FileStorageTest : public ::testing::Test {
         FLAGS_logtostderr = true;
         UnsetEnv("MOONCAKE_OFFLOAD_FILE_STORAGE_PATH");
         UnsetEnv("MOONCAKE_OFFLOAD_LOCAL_BUFFER_SIZE_BYTES");
-        UnsetEnv("MOONCAKE_OFFLOAD_BUCKET_ITERATOR_KEYS_LIMIT");
+        UnsetEnv("MOONCAKE_OFFLOAD_SCANMETA_ITERATOR_KEYS_LIMIT");
+        UnsetEnv("MOONCAKE_SCANMETA_ITERATOR_KEYS_LIMIT");
         UnsetEnv("MOONCAKE_OFFLOAD_BUCKET_KEYS_LIMIT");
         UnsetEnv("MOONCAKE_OFFLOAD_BUCKET_SIZE_LIMIT_BYTES");
         UnsetEnv("MOONCAKE_OFFLOAD_TOTAL_KEYS_LIMIT");

mooncake-store/tests/offload_on_evict_test.cpp

@@ -0,0 +1,388 @@
+#include "master_service.h"
+
+#include <glog/logging.h>
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <chrono>
+#include <memory>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "types.h"
+
+namespace mooncake::test {
+
+class OffloadOnEvictTest : public ::testing::Test {
+   protected:
+    void SetUp() override {
+        google::InitGoogleLogging("OffloadOnEvictTest");
+        FLAGS_logtostderr = true;
+    }
+
+    void TearDown() override { google::ShutdownGoogleLogging(); }
+
+    static constexpr size_t kDefaultSegmentBase = 0x300000000;
+
+    Segment MakeSegment(std::string name, size_t base, size_t size) const {
+        Segment segment;
+        segment.id = generate_uuid();
+        segment.name = std::move(name);
+        segment.base = base;
+        segment.size = size;
+        segment.te_endpoint = segment.name;
+        return segment;
+    }
+
+    struct MountedSegmentContext {
+        UUID segment_id;
+        UUID client_id;
+    };
+
+    MountedSegmentContext PrepareSegment(MasterService& service,
+                                         std::string name, size_t base,
+                                         size_t size) const {
+        Segment segment = MakeSegment(std::move(name), base, size);
+        UUID client_id = generate_uuid();
+        auto mount_result = service.MountSegment(segment, client_id);
+        EXPECT_TRUE(mount_result.has_value());
+        return {.segment_id = segment.id, .client_id = client_id};
+    }
+
+    // Put an object and complete it.
+    void PutObject(MasterService& service, const UUID& client_id,
+                   const std::string& key, size_t size = 1024) {
+        ReplicateConfig config;
+        config.replica_num = 1;
+        auto put_start = service.PutStart(client_id, key, size, config);
+        ASSERT_TRUE(put_start.has_value()) << "PutStart failed for key=" << key;
+        auto put_end = service.PutEnd(client_id, key, ReplicaType::MEMORY);
+        ASSERT_TRUE(put_end.has_value()) << "PutEnd failed for key=" << key;
+    }
+
+    // Drain the offload queue via OffloadObjectHeartbeat.
+    std::unordered_map<std::string, int64_t> DrainOffloadQueue(
+        MasterService& service, const UUID& client_id) {
+        auto res = service.OffloadObjectHeartbeat(client_id, true);
+        if (!res) {
+            return {};
+        }
+        return std::move(res.value());
+    }
+
+    template <typename Predicate>
+    void WaitUntil(
+        Predicate&& predicate,
+        std::chrono::milliseconds timeout = std::chrono::milliseconds(4000),
+        std::chrono::milliseconds interval =
+            std::chrono::milliseconds(50)) const {
+        const auto deadline = std::chrono::steady_clock::now() + timeout;
+        while (std::chrono::steady_clock::now() < deadline) {
+            if (predicate()) {
+                return;
+            }
+            std::this_thread::sleep_for(interval);
+        }
+        EXPECT_TRUE(predicate());
+    }
+
+    // Fill a segment until PutStart fails, triggering eviction.
+    // Returns the number of successful puts.
+    int FillSegmentUntilEviction(MasterService& service, const UUID& client_id,
+                                 const std::string& key_prefix,
+                                 size_t object_size, int max_puts) {
+        int success_puts = 0;
+        for (int i = 0; i < max_puts; ++i) {
+            std::string key = key_prefix + std::to_string(i);
+            ReplicateConfig config;
+            config.replica_num = 1;
+            auto result = service.PutStart(client_id, key, object_size, config);
+            if (result.has_value()) {
+                auto end = service.PutEnd(client_id, key, ReplicaType::MEMORY);
+                EXPECT_TRUE(end.has_value());
+                success_puts++;
+            } else {
+                // Wait for eviction to process
+                std::this_thread::sleep_for(std::chrono::milliseconds(50));
+            }
+        }
+        return success_puts;
+    }
+};
+
+// =============================================================================
+// Combo A: Default config (offload at PutEnd)
+// =============================================================================
+
+TEST_F(OffloadOnEvictTest, ComboA_OffloadAtPutEnd) {
+    MasterServiceConfig config;
+    config.enable_offload = true;
+    config.default_kv_lease_ttl = 2000;
+    auto service = std::make_unique<MasterService>(config);
+
+    constexpr size_t seg_size = 1024 * 1024 * 16;
+    auto ctx =
+        PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
+
+    // Mount local disk segment with offloading ENABLED
+    auto mount_ld = service->MountLocalDiskSegment(ctx.client_id, true);
+    ASSERT_TRUE(mount_ld.has_value());
+
+    // Put objects
+    PutObject(*service, ctx.client_id, "key_a1");
+    PutObject(*service, ctx.client_id, "key_a2");
+    PutObject(*service, ctx.client_id, "key_a3");
+
+    // Default mode: PutEnd pushes to offload queue immediately
+    auto queued = DrainOffloadQueue(*service, ctx.client_id);
+    EXPECT_EQ(queued.size(), 3u)
+        << "Default: all 3 objects should be in offload queue after PutEnd";
+    EXPECT_TRUE(queued.count("key_a1"));
+    EXPECT_TRUE(queued.count("key_a2"));
+    EXPECT_TRUE(queued.count("key_a3"));
+
+    service->RemoveAll();
+}
+
+TEST_F(OffloadOnEvictTest, ComboA_EvictionWorks) {
+    // Regression: eviction still works in default mode
+    const uint64_t kv_lease_ttl = 2000;
+    MasterServiceConfig config;
+    config.enable_offload = true;
+    config.default_kv_lease_ttl = kv_lease_ttl;
+    auto service = std::make_unique<MasterService>(config);
+
+    // Large segment: can hold ~16K objects of 15KB
+    constexpr size_t seg_size = 1024 * 1024 * 16 * 15;
+    constexpr size_t object_size = 1024 * 15;
+    auto ctx =
+        PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
+
+    // Put more objects than the segment can hold
+    int success_puts = FillSegmentUntilEviction(
+        *service, ctx.client_id, "evict_a_", object_size, 1024 * 16 + 50);
+    EXPECT_GT(success_puts, 1024 * 16)
+        << "Default: eviction should allow more puts than capacity";
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(kv_lease_ttl));
+    service->RemoveAll();
+}
+
+// =============================================================================
+// Combo B: offload_on_evict=true (offload on evict, no force-evict)
+// =============================================================================
+
+TEST_F(OffloadOnEvictTest, ComboB_PutEndSkipsOffloadQueue) {
+    MasterServiceConfig config;
+    config.enable_offload = true;
+    config.offload_on_evict = true;
+    config.default_kv_lease_ttl = 2000;
+    auto service = std::make_unique<MasterService>(config);
+
+    constexpr size_t seg_size = 1024 * 1024 * 16;
+    auto ctx =
+        PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
+    auto mount_ld = service->MountLocalDiskSegment(ctx.client_id, true);
+    ASSERT_TRUE(mount_ld.has_value());
+
+    PutObject(*service, ctx.client_id, "key_b1");
+    PutObject(*service, ctx.client_id, "key_b2");
+    PutObject(*service, ctx.client_id, "key_b3");
+
+    // Offload-on-evict: PutEnd should NOT push to offload queue
+    auto queued = DrainOffloadQueue(*service, ctx.client_id);
+    EXPECT_EQ(queued.size(), 0u)
+        << "Offload-on-evict: queue should be empty after PutEnd";
+
+    service->RemoveAll();
+}
+
+TEST_F(OffloadOnEvictTest, ComboB_EvictionTriggersOffload) {
+    const uint64_t kv_lease_ttl = 2000;
+    MasterServiceConfig config;
+    config.enable_offload = true;
+    config.offload_on_evict = true;
+    config.default_kv_lease_ttl = kv_lease_ttl;
+    auto service = std::make_unique<MasterService>(config);
+
+    constexpr size_t seg_size = 1024 * 1024 * 16 * 15;
+    constexpr size_t object_size = 1024 * 15;
+    auto ctx =
+        PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
+    auto mount_ld = service->MountLocalDiskSegment(ctx.client_id, true);
+    ASSERT_TRUE(mount_ld.has_value());
+
+    // Fill segment to trigger eviction
+    bool eviction_triggered = false;
+    int success_puts = 0;
+    for (int i = 0; i < 1024 * 16 + 50; ++i) {
+        std::string key = "evict_b_" + std::to_string(i);
+        ReplicateConfig config;
+        config.replica_num = 1;
+        auto result =
+            service->PutStart(ctx.client_id, key, object_size, config);
+        if (result.has_value()) {
+            auto end = service->PutEnd(ctx.client_id, key, ReplicaType::MEMORY);
+            ASSERT_TRUE(end.has_value());
+            success_puts++;
+        } else {
+            eviction_triggered = true;
+            std::this_thread::sleep_for(std::chrono::milliseconds(50));
+        }
+    }
+
+    EXPECT_TRUE(eviction_triggered)
+        << "Eviction should trigger when segment fills up";
+
+    // Offload-on-evict: eviction should push objects to offload queue
+    auto queued = DrainOffloadQueue(*service, ctx.client_id);
+    EXPECT_GT(queued.size(), 0u)
+        << "Offload-on-evict: eviction should push to offload queue";
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(kv_lease_ttl));
+    service->RemoveAll();
+}
+
+TEST_F(OffloadOnEvictTest, ComboB_NoFallbackWithoutForceEvict) {
+    // Without force_evict AND without a LocalDiskSegment, offload queue push
+    // fails and eviction does NOT force-delete MEMORY (data-preserving).
+    // The segment fills and subsequent puts fail — this is the safe default.
+    const uint64_t kv_lease_ttl = 2000;
+    MasterServiceConfig config;
+    config.enable_offload = true;
+    config.offload_on_evict = true;
+    config.default_kv_lease_ttl = kv_lease_ttl;
+    auto service = std::make_unique<MasterService>(config);
+
+    // NO local disk segment mounted — PushOffloadingQueue will fail
+    constexpr size_t seg_size = 1024 * 1024 * 16 * 15;
+    constexpr size_t object_size = 1024 * 15;
+    auto ctx =
+        PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
+
+    // Without force_evict, push failures mean DRAM cannot be freed,
+    // so we can only put up to segment capacity (no overflow).
+    int success_puts = FillSegmentUntilEviction(
+        *service, ctx.client_id, "evict_b2_", object_size, 1024 * 16 + 50);
+    EXPECT_LE(success_puts, 1024 * 16)
+        << "Without force_evict, segment should fill and stay full";
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(kv_lease_ttl));
+    service->RemoveAll();
+}
+
+// =============================================================================
+// Combo C: offload_on_evict=true + offload_force_evict=true
+// =============================================================================
+
+TEST_F(OffloadOnEvictTest, ComboC_PutEndSkipsOffloadQueue) {
+    MasterServiceConfig config;
+    config.enable_offload = true;
+    config.offload_on_evict = true;
+    config.offload_force_evict = true;
+    config.default_kv_lease_ttl = 2000;
+    auto service = std::make_unique<MasterService>(config);
+
+    constexpr size_t seg_size = 1024 * 1024 * 16;
+    auto ctx =
+        PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
+    auto mount_ld = service->MountLocalDiskSegment(ctx.client_id, true);
+    ASSERT_TRUE(mount_ld.has_value());
+
+    PutObject(*service, ctx.client_id, "key_c1");
+    PutObject(*service, ctx.client_id, "key_c2");
+
+    // Same as Combo B: PutEnd should skip offload queue
+    auto queued = DrainOffloadQueue(*service, ctx.client_id);
+    EXPECT_EQ(queued.size(), 0u)
+        << "Combo C: offload queue should be empty after PutEnd";
+
+    service->RemoveAll();
+}
+
+TEST_F(OffloadOnEvictTest, ComboC_EvictionWithForceEvict) {
+    const uint64_t kv_lease_ttl = 2000;
+    MasterServiceConfig config;
+    config.enable_offload = true;
+    config.offload_on_evict = true;
+    config.offload_force_evict = true;
+    config.default_kv_lease_ttl = kv_lease_ttl;
+    auto service = std::make_unique<MasterService>(config);
+
+    constexpr size_t seg_size = 1024 * 1024 * 16 * 15;
+    constexpr size_t object_size = 1024 * 15;
+    auto ctx =
+        PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
+    auto mount_ld = service->MountLocalDiskSegment(ctx.client_id, true);
+    ASSERT_TRUE(mount_ld.has_value());
+
+    // With force-evict, eviction should work effectively.
+    // Note: without a real FileStorage heartbeat, offloaded objects' refcnt
+    // never decreases, so DRAM isn't fully freed beyond what direct eviction
+    // allows. We verify eviction doesn't deadlock (can fill to capacity).
+    int success_puts = FillSegmentUntilEviction(
+        *service, ctx.client_id, "evict_c_", object_size, 1024 * 16 + 50);
+    EXPECT_GE(success_puts, 1024 * 16)
+        << "Combo C: eviction should work with force-evict enabled";
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(kv_lease_ttl));
+    service->RemoveAll();
+}
+
+// =============================================================================
+// Combo D: offload_force_evict=true only (should be no-op without on_evict)
+// =============================================================================
+
+TEST_F(OffloadOnEvictTest, ComboD_ForceEvictAloneIsIgnored) {
+    MasterServiceConfig config;
+    config.enable_offload = true;
+    config.offload_force_evict = true;  // on_evict is false → force is ignored
+    config.default_kv_lease_ttl = 2000;
+    auto service = std::make_unique<MasterService>(config);
+
+    constexpr size_t seg_size = 1024 * 1024 * 16;
+    auto ctx =
+        PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
+    auto mount_ld = service->MountLocalDiskSegment(ctx.client_id, true);
+    ASSERT_TRUE(mount_ld.has_value());
+
+    // Should behave like Combo A (default: offload at PutEnd)
+    PutObject(*service, ctx.client_id, "key_d1");
+    PutObject(*service, ctx.client_id, "key_d2");
+
+    auto queued = DrainOffloadQueue(*service, ctx.client_id);
+    EXPECT_EQ(queued.size(), 2u)
+        << "Combo D: FORCE_EVICT alone should not change default behavior";
+
+    service->RemoveAll();
+}
+
+TEST_F(OffloadOnEvictTest, ComboD_EvictionWorks) {
+    const uint64_t kv_lease_ttl = 2000;
+    MasterServiceConfig config;
+    config.enable_offload = true;
+    config.offload_force_evict = true;  // on_evict is false → force is ignored
+    config.default_kv_lease_ttl = kv_lease_ttl;
+    auto service = std::make_unique<MasterService>(config);
+
+    constexpr size_t seg_size = 1024 * 1024 * 16 * 15;
+    constexpr size_t object_size = 1024 * 15;
+    auto ctx =
+        PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
+
+    int success_puts = FillSegmentUntilEviction(
+        *service, ctx.client_id, "evict_d_", object_size, 1024 * 16 + 50);
+    EXPECT_GT(success_puts, 1024 * 16)
+        << "Combo D: eviction should work normally";
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(kv_lease_ttl));
+    service->RemoveAll();
+}
+
+}  // namespace mooncake::test
+
+int main(int argc, char** argv) {
+    ::testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}

mooncake-transfer-engine/CMakeLists.txt

@@ -9,10 +9,11 @@ if (NOT GLOBAL_CONFIG)
   include(../mooncake-common/FindGLOG.cmake)
   include(../mooncake-common/common.cmake)
 
-  # Build asio_shared library for standalone build
-  # Set compile definitions before add_subdirectory to ensure proper compilation
+  # Build asio_shared in mooncake-common for standalone builds.
+  # Set compile definitions before adding targets that include ASIO headers.
   add_compile_definitions(ASIO_SEPARATE_COMPILATION ASIO_DYN_LINK)
-  add_subdirectory(../mooncake-asio ${CMAKE_CURRENT_BINARY_DIR}/mooncake-asio)
+  add_subdirectory(../mooncake-common/src
+                   ${CMAKE_CURRENT_BINARY_DIR}/mooncake-common-src)
 endif() # GLOBAL_CONFIG
 
 if (USE_ASCEND)

mooncake-transfer-engine/benchmark/CMakeLists.txt

@@ -17,4 +17,4 @@ endif()
 # Set RPATH for finding libasio.so at runtime
 set_target_properties(
   tebench PROPERTIES BUILD_WITH_INSTALL_RPATH TRUE
-                     INSTALL_RPATH "$ORIGIN/../lib:$ORIGIN/../../mooncake-asio")
+                     INSTALL_RPATH "$ORIGIN/../lib:$ORIGIN/../../mooncake-common")

mooncake-transfer-engine/include/multi_transport.h

@@ -55,6 +55,14 @@ class MultiTransport {
 
     Transport *getTransport(const std::string &proto);
 
+    /**
+     * @brief Check if TCP is the only installed transport.
+     *
+     * When only TCP transport is available (no RDMA, NVLink, etc.),
+     * local memcpy is preferred over TCP loopback for same-host transfers.
+     */
+    bool isTcpOnly() const;
+
     std::vector<Transport *> listTransports();
 
     void *getBaseAddr();

mooncake-transfer-engine/include/transfer_engine.h

@@ -150,6 +150,14 @@ class TransferEngine {
 
     Transport* getTransport(const std::string& proto);
 
+    /**
+     * @brief Check if TCP is the only installed transport.
+     *
+     * When only TCP transport is available (no RDMA, NVLink, etc.),
+     * local memcpy is preferred over TCP loopback for same-host transfers.
+     */
+    bool isTcpOnly() const;
+
     int syncSegmentCache(const std::string& segment_name = "");
 
     std::shared_ptr<TransferMetadata> getMetadata();

mooncake-transfer-engine/include/transfer_engine_impl.h

@@ -334,6 +334,8 @@ class TransferEngineImpl {
         return multi_transports_->getTransport(proto);
     }
 
+    bool isTcpOnly() const { return multi_transports_->isTcpOnly(); }
+
     int syncSegmentCache(const std::string& segment_name = "") {
         return metadata_->syncSegmentCache(segment_name);
     }

mooncake-transfer-engine/src/multi_transport.cpp

@@ -461,6 +461,10 @@ Transport* MultiTransport::getTransport(const std::string& proto) {
     return transport_map_[proto].get();
 }
 
+bool MultiTransport::isTcpOnly() const {
+    return transport_map_.size() == 1 && transport_map_.count("tcp") == 1;
+}
+
 std::vector<Transport*> MultiTransport::listTransports() {
     std::vector<Transport*> transport_list;
     for (auto& entry : transport_map_)

mooncake-transfer-engine/src/transfer_engine.cpp

@@ -179,6 +179,8 @@ Transport* TransferEngine::getTransport(const std::string& proto) {
     return impl_->getTransport(proto);
 }
 
+bool TransferEngine::isTcpOnly() const { return impl_->isTcpOnly(); }
+
 int TransferEngine::syncSegmentCache(const std::string& segment_name) {
     return impl_->syncSegmentCache(segment_name);
 }
@@ -578,6 +580,15 @@ Transport* TransferEngine::getTransport(const std::string& proto) {
         return impl_->getTransport(proto);
 }
 
+bool TransferEngine::isTcpOnly() const {
+    if (use_tent_)
+        // TENT already rejects TCP loopback transfers when MC_STORE_MEMCPY
+        // is disabled, so auto-enabling memcpy is unnecessary in TENT mode.
+        return false;
+    else
+        return impl_->isTcpOnly();
+}
+
 int TransferEngine::syncSegmentCache(const std::string& segment_name) {
     if (use_tent_)
         return 0;

mooncake-transfer-engine/src/transport/rdma_transport/worker_pool.cpp

@@ -161,9 +161,8 @@ int WorkerPool::submitPostSend(
         slice_queue_lock_[shard_id].unlock();
     }
 
-    submitted_slice_count_.fetch_add(submitted_slice_count,
-                                     std::memory_order_relaxed);
-    if (suspended_flag_.load(std::memory_order_relaxed)) {
+    submitted_slice_count_.fetch_add(submitted_slice_count);
+    if (suspended_flag_.load()) {
         std::lock_guard<std::mutex> lock(cond_mutex_);
         cond_var_.notify_all();
     }
@@ -398,7 +397,7 @@ void WorkerPool::transferWorker(int thread_id) {
                 // Double-check condition after acquiring lock to avoid lost
                 // wakeup
                 if (processed_slice_count_.load(std::memory_order_relaxed) ==
-                    submitted_slice_count_.load(std::memory_order_relaxed)) {
+                    submitted_slice_count_.load()) {
                     cond_var_.wait_for(lock, std::chrono::seconds(1));
                 }
                 suspended_flag_.fetch_sub(1);

mooncake-transfer-engine/tent/src/metrics/CMakeLists.txt

@@ -7,7 +7,7 @@ add_library(tent_metrics STATIC ${TENT_METRICS_SOURCES})
 # ODR safety: yalantinglibs bundles ASIO headers but does NOT compile ASIO inline because
 # ASIO_SEPARATE_COMPILATION is set globally. All ASIO symbols live exclusively in asio_shared.so,
 # so there is no risk of duplicate symbols between yalantinglibs and the rest of TE.
-# (See mooncake-asio/CMakeLists.txt: "Build asio as a shared library to avoid ODR violations")
+# (See mooncake-common/src/CMakeLists.txt: "Build asio as a shared library to avoid ODR violations")
 target_link_libraries(tent_metrics PUBLIC tent_common tent_interface yalantinglibs::yalantinglibs glog pthread)
 
 # Pass compile definition based on option
@@ -17,4 +17,4 @@ if(TENT_METRICS_ENABLED)
 else()
     target_compile_definitions(tent_metrics PUBLIC TENT_METRICS_ENABLED=0)
     message(STATUS "TENT metrics: DISABLED (zero overhead)")
-endif()
+endif()

mooncake-wheel/mooncake/mooncake_connector_v1.py

@@ -28,6 +28,19 @@ from vllm.attention.selector import get_attn_backend
 from vllm.config import VllmConfig
 from vllm.distributed.kv_transfer.kv_connector.v1.base import (
     KVConnectorBase_V1, KVConnectorMetadata, KVConnectorRole)
+
+# SupportsHMA was introduced in vllm-project/vllm PR #25712 and is enforced
+# for KV connectors by PR #27592. It is the marker the Hybrid Memory
+# Allocator uses to allow PD-disaggregation with hybrid (e.g. attention +
+# Mamba2) models. We import it conditionally so this connector keeps
+# working on older vLLM releases that pre-date the interface; on those
+# releases the marker is a no-op object base and the
+# `request_finished_all_groups` shim below is dead code.
+try:
+    from vllm.distributed.kv_transfer.kv_connector.v1.base import (
+        SupportsHMA)
+except ImportError:  # pragma: no cover - older vLLM
+    SupportsHMA = object  # type: ignore[assignment, misc]
 from vllm.distributed.parallel_state import (get_tensor_model_parallel_rank,
                                              get_tp_group)
 from vllm.forward_context import ForwardContext
@@ -104,7 +117,12 @@ class MooncakeConnectorMetadata(KVConnectorMetadata):
             self.reqs_to_send[request_id] = local_block_ids
 
 
-class MooncakeConnector(KVConnectorBase_V1):
+class MooncakeConnector(KVConnectorBase_V1, SupportsHMA):
+    # Subclassing SupportsHMA gates this connector through vLLM's Hybrid
+    # Memory Allocator path, which is required to PD-disaggregate hybrid
+    # attention + Mamba2 models (e.g. nvidia/NVIDIA-Nemotron-Nano-9B-v2).
+    # On vLLM versions that pre-date PR #25712 the marker resolves to
+    # `object` and this is identical to the previous single-base class.
 
     def __init__(self, vllm_config: VllmConfig, role: KVConnectorRole):
         assert vllm_config.kv_transfer_config is not None
@@ -154,6 +172,28 @@ class MooncakeConnector(KVConnectorBase_V1):
         assert self.connector_scheduler is not None
         return self.connector_scheduler.request_finished(request, block_ids)
 
+    def request_finished_all_groups(
+        self,
+        request: "Request",
+        block_ids: tuple[list[int], ...],
+    ) -> tuple[bool, Optional[dict[str, Any]]]:
+        """SupportsHMA hook for hybrid (multi-group) KV cache layouts.
+
+        Hybrid models (e.g. attention + Mamba2) expose one block-id list
+        per KV cache group instead of a single flat list. The Mooncake
+        transport itself does not yet distinguish groups on the wire, so
+        we flatten the per-group lists and delegate to the existing
+        single-group `request_finished`. This is the minimum-viable shim
+        that satisfies the `SupportsHMA` contract and unblocks the
+        Hybrid Memory Allocator gate so the engine can start up; cross-
+        node fidelity for non-attention SSM/Mamba state is not asserted
+        by this method and remains a follow-up (the Mamba2 backend in
+        vLLM still raises NotImplementedError from
+        `get_kv_cache_shape()`).
+        """
+        flat: list[int] = [b for group in block_ids for b in group]
+        return self.request_finished(request, flat)
+
     ############################################################
     # Worker Side Methods
     ############################################################

mooncake-wheel/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "mooncake-transfer-engine"
-version = "0.3.10.post1"
+version = "0.3.10.post2"
 description = "Python binding of a Mooncake library using pybind11"
 authors = [
     { name = "Mooncake Authors" }

mooncake-wheel/tests/test_offload_on_eviction.py

@@ -0,0 +1,144 @@
+"""Python-binding test for the offload-on-evict feature.
+
+Unlike the default ``enable_ssd_offload`` path where PutEnd synchronously
+queues objects for offload, the offload-on-evict mode defers offload until
+DRAM eviction triggers. This test asserts the core behavioral difference:
+small workloads below the eviction watermark stay in DRAM only and do not
+produce LOCAL_DISK replicas.
+
+The end-to-end overflow → evict → offload → read contract is covered by
+``test_ssd_offload_in_evict.py`` (legacy sync-offload mode) and the
+``verify_offload_on_eviction`` scenario in the kv-cache-offload-benchmark
+project, both of which exercise the same storage backend code paths as
+this feature.
+
+Prerequisites:
+  - mooncake_master running with master.json containing:
+      ``"enable_offload": true``
+      ``"offload_on_evict": true``
+  - ``MOONCAKE_OFFLOAD_FILE_STORAGE_PATH=/tmp/...`` env var set on the
+    client side so FileStorage uses the matching backend / path.
+"""
+
+import unittest
+import os
+import time
+from mooncake.store import MooncakeDistributedStore
+
+# Align with the master's default_kv_lease_ttl (override via env).
+DEFAULT_DEFAULT_KV_LEASE_TTL = 5000  # ms
+default_kv_lease_ttl = int(
+    os.getenv("DEFAULT_KV_LEASE_TTL", DEFAULT_DEFAULT_KV_LEASE_TTL)
+)
+
+SEGMENT_SIZE = int(os.getenv("SEGMENT_SIZE_BYTES", str(128 * 1024 * 1024)))
+LOCAL_BUFFER_SIZE = int(os.getenv("LOCAL_BUFFER_SIZE_BYTES", str(128 * 1024 * 1024)))
+
+
+def get_client(store):
+    """Initialize and setup the distributed store client."""
+    protocol = os.getenv("PROTOCOL", "tcp")
+    device_name = os.getenv("DEVICE_NAME", "")
+    local_hostname = os.getenv("LOCAL_HOSTNAME", "localhost")
+    metadata_server = os.getenv("MC_METADATA_SERVER", "P2PHANDSHAKE")
+    master_server_address = os.getenv("MASTER_SERVER", "127.0.0.1:50051")
+
+    retcode = store.setup(
+        local_hostname,
+        metadata_server,
+        SEGMENT_SIZE,
+        LOCAL_BUFFER_SIZE,
+        protocol,
+        device_name,
+        master_server_address,
+        None,  # engine
+        True,  # enable_ssd_offload — required for FileStorage / LOCAL_DISK
+    )
+    if retcode:
+        raise RuntimeError(f"Failed to setup store client. Return code: {retcode}")
+
+
+class TestOffloadOnEviction(unittest.TestCase):
+    """Python-binding test for offload-on-evict behavioral difference."""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.store = MooncakeDistributedStore()
+        get_client(cls.store)
+
+    def test_small_workload_no_disk_replica(self):
+        """Workload well below the eviction watermark must NOT produce
+        LOCAL_DISK replicas.
+
+        Under the default ``enable_ssd_offload=true`` path, every PutEnd
+        would queue the key for offload (heartbeat → disk). Under
+        offload-on-evict, the offload queue stays empty until DRAM
+        pressure triggers eviction. This asymmetry is what the feature
+        is for — avoiding redundant disk I/O for hot data.
+
+        A zero disk-replica count thus doubles as a guard that the
+        master is actually running with ``offload_on_evict: true`` in config.
+        """
+        VALUE_SIZE = 1024  # 1 KB
+        NUM_KEYS = 16  # 16 KB total, several orders of magnitude below
+        # the 128 MB segment — eviction cannot trigger.
+
+        timestamp = int(time.time())
+        keys = [f"ooe_small_{i}_{timestamp}" for i in range(NUM_KEYS)]
+        reference = {}
+
+        try:
+            for key in keys:
+                value = os.urandom(VALUE_SIZE)
+                retcode = self.store.put(key, value)
+                self.assertEqual(
+                    retcode,
+                    0,
+                    f"Put failed for {key}: retcode={retcode}",
+                )
+                reference[key] = value
+
+            # Wait past one heartbeat interval so any async offload would
+            # have had a chance to complete.
+            time.sleep(3)
+
+            # Query replicas — they must all be MEMORY only.
+            descs = self.store.batch_get_replica_desc(keys)
+            disk_replicas = 0
+            for key in keys:
+                infos = descs.get(key) if isinstance(descs, dict) else None
+                if infos is None:
+                    continue
+                if not isinstance(infos, (list, tuple)):
+                    infos = [infos]
+                for info in infos:
+                    type_str = str(getattr(info, "type", ""))
+                    if type_str and "MEMORY" not in type_str:
+                        disk_replicas += 1
+
+            self.assertEqual(
+                disk_replicas,
+                0,
+                f"Offload-on-evict should keep small workloads in DRAM "
+                f"only, but found {disk_replicas} LOCAL_DISK replicas. "
+                f"Is offload_on_evict=true set in master config?",
+            )
+
+            # Sanity: every key is readable and bit-exact (served from DRAM).
+            for key, expected in reference.items():
+                self.assertEqual(
+                    self.store.get(key), expected, f"Bit-exact mismatch for {key}"
+                )
+        finally:
+            # Cleanup — remove all keys written by this test regardless of
+            # success or failure so subsequent tests start with clean DRAM.
+            for key in keys:
+                try:
+                    self.store.remove(key)
+                except Exception:
+                    pass
+            time.sleep(default_kv_lease_ttl / 1000 + 0.5)
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/build_wheel.sh

@@ -16,7 +16,7 @@ BUILD_DIR="${BUILD_DIR:-build}"
 echo "Building wheel for Python ${PYTHON_VERSION} with output directory ${OUTPUT_DIR}"
 
 # Ensure LD_LIBRARY_PATH includes /usr/local/lib
-export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/$(pwd)/build/mooncake-asio:/usr/local/lib
+export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/$(pwd)/build/mooncake-common:/usr/local/lib
 
 echo "Cleaning wheel-build directory"
 rm -rf mooncake-wheel/mooncake_transfer_engine*
@@ -29,7 +29,7 @@ echo "Creating directory structure..."
 cp build/mooncake-integration/engine.*.so mooncake-wheel/mooncake/engine.so
 
 # Copy libasio.so to mooncake directory (runtime dependency of engine.so)
-cp build/mooncake-asio/libasio.so mooncake-wheel/mooncake/libasio.so
+cp build/mooncake-common/libasio.so mooncake-wheel/mooncake/libasio.so
 
 # Copy store.so to mooncake directory
 if [ -f build/mooncake-integration/store.*.so ]; then
@@ -170,11 +170,8 @@ rm -rf ${OUTPUT_DIR}/
 mkdir -p ${OUTPUT_DIR}
 
 echo "Installing required build packages"
-if command -v python${PYTHON_VERSION} &>/dev/null; then
+if command -v pip &>/dev/null; then
     python${PYTHON_VERSION} -m pip install --upgrade pip build setuptools wheel auditwheel
-elif command -v pip &>/dev/null; then
-    pip install --upgrade pip
-    pip install build setuptools wheel auditwheel
 elif command -v uv &>/dev/null; then
     uv pip install --upgrade pip
     uv pip install build setuptools wheel auditwheel