feat(umbp): tunable read leases and tighter master lease default

src/umbp/distributed/bin/master_main.cpp

@@ -48,11 +48,8 @@ int main(int argc, char** argv) {
 
   MORI_UMBP_INFO(
       "[Master] Resolved timing: heartbeat_ttl={}s reaper_interval={}s "
-      "allocation_ttl={}s finalized_record_ttl={}s max_missed={} "
-      "eviction.check_interval={}s lease_duration={}s",
+      "max_missed={} eviction.check_interval={}s lease_duration={}s",
       config.registry_config.heartbeat_ttl.count(), config.registry_config.reaper_interval.count(),
-      config.registry_config.allocation_ttl.count(),
-      config.registry_config.finalized_record_ttl.count(),
       config.registry_config.max_missed_heartbeats, config.eviction_config.check_interval.count(),
       config.eviction_config.lease_duration.count());
 

src/umbp/distributed/master/master_client.cpp

@@ -382,15 +382,17 @@ namespace {
 // Auto-flush the heartbeat once this many KvEvents accumulate at the peer, so a
 // completed batch of puts becomes visible at the master without the application
 // calling FlushHeartbeat().  Defaults to one typical put batch (128 keys).  An
-// unset / unparseable / zero value falls back to the default; to make auto-flush
-// effectively never trigger, set it to a very large number.
+// unset / unparseable value falls back to the default.  An explicit `0`
+// disables auto-flush entirely (ADDs then only ship on the heartbeat interval
+// or an explicit Flush()); to keep auto-flush enabled but effectively never
+// trigger on size, set it to a very large number instead.
 size_t AutoFlushEventThreshold() {
   static const size_t v = [] {
     const char* e = std::getenv("UMBP_AUTO_FLUSH_EVENT_THRESHOLD");
     if (e == nullptr) return size_t{128};
     char* end = nullptr;
     unsigned long long n = std::strtoull(e, &end, 10);
-    if (end == e || n == 0) return size_t{128};
+    if (end == e) return size_t{128};  // unparseable -> default; 0 passes through as "disabled"
     return static_cast<size_t>(n);
   }();
   return v;

src/umbp/distributed/peer/peer_dram_allocator.cpp

@@ -473,8 +473,10 @@ void PeerDramAllocator::QueueEventLocked(KvEvent event) {
   // are appended one at a time, so `==` fires at most once per batch.  Called
   // under mutex_: auto_flush_cb_ must be cheap and MUST NOT re-enter the
   // allocator (it only signals the heartbeat thread).  Exactness isn't required —
-  // the heartbeat interval is the backstop.
-  if (pending_events_.size() == auto_flush_threshold_ && auto_flush_cb_) {
+  // the heartbeat interval is the backstop.  threshold_ == 0 disables size-based
+  // auto-flush entirely (ADDs then ship only on the interval / explicit Flush).
+  if (auto_flush_threshold_ > 0 && pending_events_.size() == auto_flush_threshold_ &&
+      auto_flush_cb_) {
     auto_flush_cb_();
   }
 }

src/umbp/distributed/peer/peer_service.cpp

@@ -68,7 +68,7 @@ struct StagingSlot {
 // Reclaim TTL-expired leased slots, then claim a free one as Preparing (TTL not
 // yet started) and return its index, or -1 if none free.
 int ClaimStagingSlot(std::vector<StagingSlot>& slots, std::atomic<uint64_t>& next_lease_id,
-                     std::chrono::seconds lease_timeout, StagingMetrics& metrics) {
+                     std::chrono::milliseconds lease_timeout, StagingMetrics& metrics) {
   auto now = std::chrono::steady_clock::now();
   for (auto& slot : slots) {
     if (slot.state == SlotState::kLeased && now - slot.leased_at > lease_timeout) {
@@ -173,8 +173,8 @@ class PeerServiceServer::UMBPPeerServiceImpl final : public ::umbp::UMBPPeer::Se
                       const std::vector<uint8_t>& ssd_staging_mem_desc_bytes,
                       PeerSsdManager* peer_ssd, PeerDramAllocator* dram_alloc,
                       MasterClient* master_client, StagingMetrics& metrics, int num_read_slots,
-                      int lease_timeout_s, const std::vector<uint8_t>& engine_desc_bytes,
-                      SsdCopyPipeline* copy_pipeline)
+                      std::chrono::milliseconds lease_timeout,
+                      const std::vector<uint8_t>& engine_desc_bytes, SsdCopyPipeline* copy_pipeline)
       : ssd_staging_base_(ssd_staging_base),
         ssd_staging_size_(ssd_staging_size),
         ssd_staging_mem_desc_bytes_(ssd_staging_mem_desc_bytes),
@@ -183,7 +183,7 @@ class PeerServiceServer::UMBPPeerServiceImpl final : public ::umbp::UMBPPeer::Se
         copy_pipeline_(copy_pipeline),
         engine_desc_bytes_(engine_desc_bytes),
         metrics_(metrics),
-        lease_timeout_(std::max(lease_timeout_s, 1)),
+        lease_timeout_(std::max(lease_timeout, std::chrono::milliseconds{1})),
         num_read_slots_(std::max(num_read_slots, 1)),
         // The whole staging buffer is the read region now: the lower half used
         // to be direct-put write staging, removed in the SSD-tier redesign and
@@ -581,7 +581,7 @@ class PeerServiceServer::UMBPPeerServiceImpl final : public ::umbp::UMBPPeer::Se
   const std::vector<uint8_t>& engine_desc_bytes_;
   StagingMetrics& metrics_;
 
-  const std::chrono::seconds lease_timeout_;
+  const std::chrono::milliseconds lease_timeout_;
   const int num_read_slots_;
   const uint64_t read_region_base_;
   const size_t read_slot_size_;
@@ -607,7 +607,7 @@ class PeerServiceServer::UMBPPeerServiceImpl final : public ::umbp::UMBPPeer::Se
 PeerServiceServer::PeerServiceServer(PeerDramAllocator* dram_alloc, PeerSsdManager* peer_ssd,
                                      void* ssd_staging_base, size_t ssd_staging_size,
                                      std::vector<uint8_t> ssd_staging_mem_desc_bytes,
-                                     int num_read_slots, int lease_timeout_s,
+                                     int num_read_slots, std::chrono::milliseconds lease_timeout,
                                      std::vector<uint8_t> engine_desc_bytes,
                                      MasterClient* master_client, SsdCopyPipeline* copy_pipeline)
     : ssd_staging_base_(ssd_staging_base),
@@ -620,8 +620,7 @@ PeerServiceServer::PeerServiceServer(PeerDramAllocator* dram_alloc, PeerSsdManag
       engine_desc_bytes_(std::move(engine_desc_bytes)) {
   service_ = std::make_unique<UMBPPeerServiceImpl>(
       ssd_staging_base_, ssd_staging_size_, ssd_staging_mem_desc_bytes_, peer_ssd_, dram_alloc_,
-      master_client_, metrics_, num_read_slots, lease_timeout_s, engine_desc_bytes_,
-      copy_pipeline_);
+      master_client_, metrics_, num_read_slots, lease_timeout, engine_desc_bytes_, copy_pipeline_);
 }
 
 PeerServiceServer::~PeerServiceServer() { Stop(); }

src/umbp/distributed/pool_client.cpp

@@ -323,6 +323,29 @@ std::chrono::milliseconds ReleaseLeaseRpcTimeout() {
   return v;
 }
 
+// Peer-side DRAM/HBM read lease: how long a single Resolve protects its key's
+// pages from concurrent local Evict, covering one RDMA read.  Only needs to
+// exceed one DRAM RDMA round trip (sub-ms), so 500 ms is already ~100x margin;
+// exposed so operators can tighten it under eviction pressure.
+std::chrono::milliseconds DramReadLeaseTtl() {
+  static const auto v = GetEnvMilliseconds("UMBP_DRAM_READ_LEASE_MS",
+                                           std::chrono::milliseconds(500), /*min_allowed=*/1);
+  return v;
+}
+
+// Peer-side SSD read-staging slot lease: how long a claimed staging slot is
+// reserved for a reader (peer reclaims by TTL if the best-effort ReleaseSsdLease
+// is lost) and, mirrored back to the reader, the validity window it anchors at
+// t_send.  Must exceed one SSD read + RDMA (slower than DRAM), but too long
+// pins one of only ~16 slots on a lost release, so 3 s balances both.  Also the
+// fallback for the PrepareSsdRead RPC deadline when UMBP_SSD_PREPARE_TIMEOUT_MS
+// is unset.
+std::chrono::milliseconds SsdReadLeaseTtl() {
+  static const auto v = GetEnvMilliseconds("UMBP_SSD_READ_LEASE_MS",
+                                           std::chrono::milliseconds(3000), /*min_allowed=*/1);
+  return v;
+}
+
 // ---------------------------------------------------------------------------
 //  Config / proto translation
 // ---------------------------------------------------------------------------
@@ -424,7 +447,8 @@ bool PoolClient::Init() {
   PeerDramAllocator::TierConfig hbm_cfg;  // HBM not currently exposed via PoolClientConfig
   peer_alloc_ =
       std::make_unique<PeerDramAllocator>(page_size, std::move(dram_cfg), std::move(hbm_cfg),
-                                          /*pending_ttl=*/std::chrono::milliseconds{30000});
+                                          /*pending_ttl=*/std::chrono::milliseconds{30000},
+                                          /*read_lease_ttl=*/DramReadLeaseTtl());
   peer_alloc_->StartReaper();
   master_client_->SetPeerDramAllocator(peer_alloc_.get());
 
@@ -476,7 +500,7 @@ bool PoolClient::Init() {
     peer_service_ = std::make_unique<PeerServiceServer>(
         peer_alloc_.get(), peer_ssd_.get(), ssd_staging_buffer_.get(),
         ssd_staging_buffer_ ? config_.ssd_staging_buffer_size : 0, ssd_staging_mem_desc_bytes_,
-        config_.ssd_staging_buffer_slots, config_.ssd_lease_timeout_s, engine_desc_bytes,
+        config_.ssd_staging_buffer_slots, SsdReadLeaseTtl(), engine_desc_bytes,
         master_client_.get(), ssd_copy_pipeline_.get());
     if (!peer_service_->Start(config_.peer_service_port)) {
       MORI_UMBP_ERROR("[PoolClient] PeerService failed to start on port {}",
@@ -2347,7 +2371,7 @@ PoolClient::SsdGetOutcome PoolClient::RemoteSsdReadOnce(PeerConnection& peer,
   const auto t_send = std::chrono::steady_clock::now();
   auto rpc_timeout = SsdPrepareRpcTimeoutOverride();
   if (rpc_timeout.count() == 0) {
-    rpc_timeout = std::chrono::seconds(std::max(config_.ssd_lease_timeout_s, 1));
+    rpc_timeout = SsdReadLeaseTtl();
   }
 
   ::umbp::PrepareSsdReadRequest req;

src/umbp/doc/design-master-control-plane.md

@@ -95,9 +95,7 @@ SGLang or vLLM worker). Each peer owns:
    events arrive on the next heartbeat and that's when the index
    shrinks.
 5. **One TTL.** Pending allocator slots TTL out at the peer
-   (`pending_ttl`). Master no longer maintains an allocation TTL —
-   `UMBP_ALLOCATION_TTL_SEC` is retained as a config knob for legacy
-   compatibility but is unused by the live path.
+   (`pending_ttl`). Master no longer maintains an allocation TTL.
 6. **SSD is a peer-owned, best-effort, eventually-consistent cold
    tier.** An SSD copy is an asynchronous replica filled by
    copy-on-commit on the owner peer after the DRAM/HBM commit succeeds;
@@ -310,8 +308,6 @@ Membership ledger + heartbeat ingestion.
 |---|---|
 | `heartbeat_ttl` | 10 s |
 | `reaper_interval` | 5 s |
-| `allocation_ttl` | 30 s (legacy; unused by live path) |
-| `finalized_record_ttl` | 120 s (legacy; unused by live path) |
 | `max_missed_heartbeats` | 3 |
 | `default_dram_page_size` | 2 MiB |
 
@@ -1130,7 +1126,7 @@ been removed or repurposed since the master-led era:
 | `Register` / `Unregister` / `BatchRegister` / `BatchUnregister` / `Lookup` RPCs | Removed. Use heartbeat events for membership; `RouteGet` for read-side existence. |
 | `Location.location_id` (opaque allocator handle minted on master) | Removed. `Location` is `(node_id, size, tier)`; pages live on the peer. |
 | `ClientRegistry::TrackKey` / `UntrackKey` (per-key ownership reverse index on master) | Removed. Per-key ownership is implicit in `GlobalBlockIndex`'s `Location.node_id`. |
-| Allocation TTL reaper on master | Removed. Pending TTL lives on `PeerDramAllocator`. `UMBP_ALLOCATION_TTL_SEC` is retained as a config knob for legacy compatibility but is not consumed on the live path. |
+| Allocation TTL reaper on master | Removed. Pending TTL lives on `PeerDramAllocator`. |
 | `client_id` / `MasterClientConfig::client_id` | Renamed to `node_id` everywhere. |
 
 Consult `runtime-env-vars.md` for the up-to-date `UMBP_*` knob list and

src/umbp/doc/runtime-env-vars.md

@@ -55,11 +55,9 @@ Read by the **master process** (`bin/master_main.cpp` via
 |---|---|---|---|
 | `UMBP_HEARTBEAT_TTL_SEC` | `10` | sec | Registry entry TTL; client is evicted if no heartbeat arrives within `heartbeat_ttl × max_missed_heartbeats`. |
 | `UMBP_REAPER_INTERVAL_SEC` | `5` | sec | Reaper wake-up period inside `ClientRegistry`. |
-| `UMBP_ALLOCATION_TTL_SEC` | `30` | sec | Legacy: pending allocation TTL on master. Unused on the live path (pending TTL now lives on `PeerDramAllocator`). Retained for back-compat. |
-| `UMBP_FINALIZED_RECORD_TTL_SEC` | `120` | sec | Legacy: finalized-allocation idempotency window on master. Unused on the live path. |
 | `UMBP_MAX_MISSED_HEARTBEATS` | `3` | count | Consecutive misses before a client is considered dead. |
 | `UMBP_EVICTION_CHECK_INTERVAL_SEC` | `5` | sec | `EvictionManager` loop period. |
-| `UMBP_LEASE_DURATION_SEC` | `10` | sec | Master-side read-lease length granted by `Router::RouteGet` to keep a key alive across the writer's RDMA round trip. Distinct from the peer's `read_lease_ttl_` (~500 ms by default), which protects against concurrent eviction during a single `ResolveKey`. |
+| `UMBP_LEASE_DURATION_SEC` | `2` | sec | Master-side read-lease length granted by `Router::RouteGet`: `IsLeased()` keys are skipped by the eviction scan, keeping a key alive from the moment the master returns its location until the reader connects to the owning peer. Only needs to cover the master→reader gRPC round trip + reach the peer (the actual RDMA transfer is covered peer-side by `UMBP_DRAM_READ_LEASE_MS`), so seconds is already generous; larger values pin actively-read (hot) keys against eviction. |
 | `UMBP_HEARTBEAT_INTERVAL_DIVISOR` | `2` | count | Recommended client heartbeat interval = `heartbeat_ttl / divisor`. `min_allowed=1` guards against div-by-zero. Read by the master and echoed in `RegisterClientResponse.heartbeat_interval_ms`. |
 | `UMBP_EVICTKEY_DEADLINE_MS` | `1000` | ms | Per-call gRPC deadline applied to outbound `EvictKey` RPCs from `MasterPeerStubPool`. |
 | `UMBP_HIT_INDEX_TTL_SEC` | `7200` | sec | External KV hit-count entry TTL. A hash with no counted match for longer than this is removed from the hit index. |
@@ -75,15 +73,17 @@ that has loaded `libmori_pybinds.so`).
 
 | Env var | Default | Unit | Description |
 |---|---|---|---|
+| `UMBP_DRAM_READ_LEASE_MS` | `500` | ms | Peer-side DRAM/HBM read lease: how long a single `PeerDramAllocator::Resolve` protects its key's pages from concurrent local `Evict`, covering one RDMA read of those pages. Only needs to exceed one DRAM RDMA round trip (sub-ms), so 500 ms is ~100x margin. Read once at `PoolClient::Init`; `min_allowed=1`. |
+| `UMBP_SSD_READ_LEASE_MS` | `3000` | ms | Peer-side SSD read-staging slot lease: how long a claimed staging slot is reserved before the peer reclaims it by TTL (the fallback when the reader's best-effort `ReleaseSsdLease` is lost), and, echoed back in `PrepareSsdReadResponse.lease_ttl_ms`, the reader's validity window anchored at `t_send`. Must exceed one SSD read + RDMA (slower than DRAM), but too long pins one of only ~16 slots on a lost release. Also the fallback for the `PrepareSsdRead` RPC deadline when `UMBP_SSD_PREPARE_TIMEOUT_MS` is unset. Read once at `PoolClient::Init`; `min_allowed=1`. |
 | `UMBP_RPC_SHUTDOWN_TIMEOUT_MS` | `3000` | ms | Deadline for `UnregisterClient` and the last `Heartbeat` in `~MasterClient`. Bounds `~MasterClient` worst-case at ≤ 2 × this value. |
 | `UMBP_GRPC_SHUTDOWN_DEADLINE_SEC` | `3` | sec | `server_->Shutdown(deadline)` budget, shared by master and peer service. |
 | `UMBP_METRICS_REPORT_INTERVAL_MS` | `1000` | ms | Cadence at which the pool client's `MasterClient` flushes buffered counters/gauges/histograms via `ReportMetrics`. |
 | `UMBP_RELEASE_LEASE_MAX_RETRIES` | `2` | count | `ReleaseSsdLease` RPC attempt cap on the SSD read path. `min_allowed=1`. |
 | `UMBP_SSD_GET_MAX_ATTEMPTS` | `1` | count | Total remote SSD get attempts per key. `1` = no retry. Only NO_SLOT and a reader-local lease expiry retry; rpc failure / NOT_FOUND do not. Raise to absorb staging-slot contention. `min_allowed=1`. |
 | `UMBP_SSD_GET_RETRY_BACKOFF_MS` | `2` | ms | Sleep between remote SSD get retries (only applied when another attempt follows). `min_allowed=1`. |
 | `UMBP_RELEASE_LEASE_TIMEOUT_MS` | `1000` | ms | Per-attempt gRPC deadline for the best-effort `ReleaseSsdLease` RPC so a slow peer can't stall the reader. `min_allowed=1`. |
-| `UMBP_SSD_PREPARE_TIMEOUT_MS` | `0` | ms | Per-call gRPC deadline for `PrepareSsdRead` so a hung/slow peer can't stall the serial batch. `0` = fall back to `ssd_lease_timeout_s` (cluster-homogeneous). A timed-out / failed prepare is a hard not-served outcome (NOT retried, and never a miss). `min_allowed=0`. |
-| `UMBP_AUTO_FLUSH_EVENT_THRESHOLD` | `128` | count | Peer-side unshipped `KvEvent` outbox size at which a completed batch of puts auto-triggers a heartbeat flush (`FlushHeartbeat`), so the ADDs become visible at the master without waiting for the heartbeat interval or an explicit `Flush()`. Counted on `PeerDramAllocator` only (SSD events still wait for the interval). Parsed via `std::strtoull` (no WARN on bad input); unset / unparseable / `0` -> default `128`; set to a very large value to make auto-flush effectively never fire. Cached on first use in `MasterClient::SetPeerDramAllocator`. |
+| `UMBP_SSD_PREPARE_TIMEOUT_MS` | `0` | ms | Per-call gRPC deadline for `PrepareSsdRead` so a hung/slow peer can't stall the serial batch. `0` = fall back to `UMBP_SSD_READ_LEASE_MS` (cluster-homogeneous). A timed-out / failed prepare is a hard not-served outcome (NOT retried, and never a miss). `min_allowed=0`. |
+| `UMBP_AUTO_FLUSH_EVENT_THRESHOLD` | `128` | count | Peer-side unshipped `KvEvent` outbox size at which a completed batch of puts auto-triggers a heartbeat flush (`FlushHeartbeat`), so the ADDs become visible at the master without waiting for the heartbeat interval or an explicit `Flush()`. Counted on `PeerDramAllocator` only (SSD events still wait for the interval). Parsed via `std::strtoull` (no WARN on bad input); unset / unparseable -> default `128`; `0` disables size-based auto-flush entirely (ADDs then ship only on the heartbeat interval or an explicit `Flush()`); set to a very large value to keep auto-flush armed but effectively never fire on size. Cached on first use in `MasterClient::SetPeerDramAllocator`. |
 | `UMBP_MASTER_INDEX_SHARDS` | `32` | count | Number of independently-locked, key-hashed shards backing the master `GlobalBlockIndex`. A heartbeat's event batch only takes the exclusive lock on the shards its keys hash into, so unrelated `RoutePut` / `BatchLookup` readers on other shards don't block behind a large apply; full-sync (`ReplaceNodeLocations`) likewise becomes N small critical sections instead of one giant one. Read once at master start via `std::strtol`; unset / unparseable / `< 1` -> default `32` (a WARN is logged on unparseable input). `1` reproduces the old single-global-lock behavior. |
 
 ## SPDK proxy