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

--------- Signed-off-by: Vladislav Nosivskoy <vladnosiv@gmail.com>
[Store] Support storage hierarchy with offload-on-evict mode (#1899 )
2026-04-22 17:53:16 +08:00 · 2026-04-22 17:48:07 +08:00 · 2026-04-22 12:54:51 +08:00 · 2026-04-22 12:49:47 +08:00 · 2026-04-22 11:01:31 +08:00 · 2026-04-22 00:15:33 +08:00
34 changed files with 2411 additions and 94 deletions
--- a/.github/workflows/ci.yml
+++ b/.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/...
--- a/.gitignore
+++ b/.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/
--- a/CMakeLists.txt
+++ b/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)
--- a/dependencies.sh
+++ b/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() {
--- a/docs/source/design/transfer-engine/efa_transport.md
+++ b/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
--- a/docs/source/design/transfer-engine/kunpeng_ub_transport.md
+++ b/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
--- a/mooncake-asio/CMakeLists.txt
+++ b/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)
--- a/mooncake-common/CMakeLists.txt
+++ b/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)

--- a/mooncake-common/etcd/CMakeLists.txt
+++ b/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
-)
+)
--- a/mooncake-common/etcd/go.mod
+++ b/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
--- a/mooncake-common/etcd/go.sum
+++ b/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=
--- a/mooncake-common/k8s-lease/CMakeLists.txt
+++ b/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
+)
--- a/mooncake-common/k8s-lease/cmd/envtest-server/main.go
+++ b/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()
+}
--- a/mooncake-common/k8s-lease/go.mod
+++ b/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
+)
--- a/mooncake-common/k8s-lease/k8s_lease_integration_test.go
+++ b/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
+}
--- a/mooncake-common/k8s-lease/k8s_lease_wrapper.go
+++ b/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() {}
--- a/mooncake-common/k8s-lease/k8s_lease_wrapper_test.go
+++ b/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)
+		}
+	}
+}
--- a/mooncake-common/src/CMakeLists.txt
+++ b/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)
--- a/mooncake-common/src/asio_impl.cpp
+++ b/mooncake-common/src/asio_impl.cpp
--- a/mooncake-p2p-store/build.sh
+++ b/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
--- a/mooncake-store/go/build.sh
+++ b/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"

--- a/mooncake-store/include/master_config.h
+++ b/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;
--- a/mooncake-store/include/master_service.h
+++ b/mooncake-store/include/master_service.h
@ -1168,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_;
--- a/mooncake-store/src/master.cpp
+++ b/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
--- a/mooncake-store/src/master_service.cpp
+++ b/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";
@ -910,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) {
@ -3491,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;
@ -3563,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);
@ -3624,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;
@ -3674,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;
@ -3704,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);
@ -3716,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() {
--- a/mooncake-store/src/real_client_main.cpp
+++ b/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.
--- a/mooncake-store/tests/CMakeLists.txt
+++ b/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)
--- a/mooncake-store/tests/offload_on_evict_test.cpp
+++ b/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();
+}
--- a/mooncake-transfer-engine/CMakeLists.txt
+++ b/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)
--- a/mooncake-transfer-engine/benchmark/CMakeLists.txt
+++ b/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")
--- a/mooncake-transfer-engine/tent/src/metrics/CMakeLists.txt
+++ b/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()
--- a/mooncake-wheel/pyproject.toml
+++ b/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" }
--- a/mooncake-wheel/tests/test_offload_on_eviction.py
+++ b/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()
--- a/scripts/build_wheel.sh
+++ b/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
Author	SHA1	Message	Date
Vladislav Nosivskoy	24e29df083	[Store][K8s-Native][1/N] Go-library for k8s leader election (#1910 ) --------- Signed-off-by: Vladislav Nosivskoy <vladnosiv@gmail.com>	2026-04-22 17:53:16 +08:00
Geonwoo Choi	db7bce3056	[Store] Support storage hierarchy with offload-on-evict mode (#1899 ) Signed-off-by: hnts03-moreh <geonwoo.choi@moreh.io> Signed-off-by: Geonwoo Shin <geonwoo.shin@moreh.io>	2026-04-22 17:48:07 +08:00
人生若只如初见	a14e0b600a	[Store] fix: setup logs dir for realclient_main (#1925 ) Signed-off-by: guoxiang <mzygdeaq@qq.com>	2026-04-22 12:54:51 +08:00
Shangming Cai	7442626169	[Build] Update build_wheel.sh package installation logic (#1950 ) Refactor package installation logic to prioritize pip command.	2026-04-22 12:49:47 +08:00
Shangming Cai	e1d6d6f6f4	Bump version to 0.3.10.post2 in pyproject.toml (#1949 )	2026-04-22 11:01:31 +08:00
Copilot	255e287bc1	Refactor ASIO shared target into mooncake-common (#1926 ) Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: stmatengss <11641725+stmatengss@users.noreply.github.com> Co-authored-by: zhang zuoyuan <zhangzuo21@mails.tsinghua.edu.cn>	2026-04-22 00:15:33 +08:00
Copilot	cfea2cb0f5	[Common][Etcd] Pin a patched Go toolchain and track etcd go.sum for wrapper builds (#1937 ) Agent-Logs-Url: https://github.com/kvcache-ai/Mooncake/sessions/a9f47095-8ff4-4b08-9e3d-abcf3e5773b5 Co-authored-by: stmatengss <11641725+stmatengss@users.noreply.github.com> Agent-Logs-Url: https://github.com/kvcache-ai/Mooncake/sessions/53426ab5-464c-4978-86b2-da9850e70aa0 Co-authored-by: stmatengss <11641725+stmatengss@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: stmatengss <11641725+stmatengss@users.noreply.github.com>	2026-04-21 22:39:48 +08:00