tl::expected<void, ErrorCode> Client::Put(const ObjectKey& key,
                                          std::vector<Slice>& slices,
                                          const ReplicateConfig& config) {
    // 1. 准备 slice lengths
    std::vector<size_t> slice_lengths;
    for (size_t i = 0; i < slices.size(); ++i) {
        slice_lengths.emplace_back(slices[i].size);
    }

    // 2. 向 Master 请求分配空间
    auto start_result = master_client_.PutStart(key, slice_lengths, config);
    // ...

auto MasterService::PutStart(const UUID& client_id, const std::string& key,
                             const uint64_t slice_length,
                             const ReplicateConfig& config)
    -> tl::expected<std::vector<Replica::Descriptor>, ErrorCode> {
    // 1. 参数校验 (L554-559)
    if (config.replica_num == 0 || key.empty() || slice_length == 0) { ... }

    // 2. 锁定 shard 并检查对象是否存在 (L577-603)
    size_t shard_idx = getShardIndex(key);
    MutexLocker lock(&metadata_shards_[shard_idx].mutex);
    // ...检查是否已存在,处理过期 PutStart...

    // 3. 分配副本空间 (L606-632)
    ScopedAllocatorAccess allocator_access = segment_manager_.getAllocatorAccess();
    auto allocation_result = allocation_strategy_->Allocate(
        allocator_manager, slice_length, config.replica_num, preferred_segments);

    // 4. 如果启用磁盘副本,分配磁盘路径 (L635-639)
    if (use_disk_replica_) {
        std::string file_path = ResolvePath(key);
        replicas.emplace_back(file_path, total_length, ReplicaStatus::PROCESSING);
    }

    // 5. 写入元数据 (L650-655)
    metadata_shards_[shard_idx].metadata.emplace(...);
    metadata_shards_[shard_idx].processing_keys.insert(key);

// 先处理磁盘副本
if (storage_backend_) {
    for (auto it = start_result.value().rbegin(); ...) {
        if (replica.is_disk_replica()) {
            auto disk_descriptor = replica.get_disk_descriptor();
            PutToLocalFile(key, slices, disk_descriptor);
            break;
        }
    }
}

for (const auto& replica : start_result.value()) {
    if (replica.is_memory_replica()) {
        ErrorCode transfer_err = TransferWrite(replica, slices);
        if (transfer_err != ErrorCode::OK) {
            // 传输失败,撤销操作
            auto revoke_result = master_client_.PutRevoke(key, ReplicaType::MEMORY);
            return tl::unexpected(transfer_err);
        }
    }
}

auto end_result = master_client_.PutEnd(key, ReplicaType::MEMORY);

tl::expected<void, ErrorCode> Client::Get(const std::string& object_key,
                                          std::vector<Slice>& slices) {
    auto query_result = Query(object_key);    // 查询元数据
    if (!query_result) {
        return tl::unexpected(query_result.error());
    }
    return Get(object_key, query_result.value(), slices);
}

tl::expected<QueryResult, ErrorCode> Client::Query(const std::string& object_key) {
    auto result = master_client_.GetReplicaList(object_key);
    // ...
    return QueryResult(std::move(result.value().replicas),
        start_time + std::chrono::milliseconds(result.value().lease_ttl_ms));
}

ErrorCode Client::FindFirstCompleteReplica(
    const std::vector<Replica::Descriptor>& replica_list,
    Replica::Descriptor& replica) {
    for (size_t i = 0; i < replica_list.size(); ++i) {
        if (replica_list[i].status == ReplicaStatus::COMPLETE) {
            replica = replica_list[i];
            return ErrorCode::OK;
        }
    }
    return ErrorCode::INVALID_REPLICA;
}

tl::expected<void, ErrorCode> Client::Get(const std::string& object_key,
                                          const QueryResult& query_result,
                                          std::vector<Slice>& slices) {
    Replica::Descriptor replica;
    ErrorCode err = FindFirstCompleteReplica(query_result.replicas, replica);
    // ...
    err = TransferRead(replica, slices);   // 发起 RDMA 读
    // ...
    if (query_result.IsLeaseExpired()) {   // 检查 lease 是否过期
        return tl::unexpected(ErrorCode::LEASE_EXPIRED);
    }
}

void FilereadWorkerPool::submitTask(FilereadTask task) {
    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        task_queue_.push(std::move(task));
    }
    queue_cv_.notify_one();
}

void FilereadWorkerPool::workerThread() {
    // ... 等待任务 ...
    auto load_result = backend_->LoadObject(
        task.file_path, task.slices, task.object_size);
}

if (query_result.IsLeaseExpired()) {
    LOG(WARNING) << "lease_expired_before_data_transfer_completed key=" << object_key;
    return tl::unexpected(ErrorCode::LEASE_EXPIRED);
}

// 在 MasterService 构造函数中启动
eviction_running_ = true;
eviction_thread_ = std::thread(&MasterService::EvictionThreadFunc, this);

void MasterService::EvictionThreadFunc() {
    while (eviction_running_) {
        double used_ratio = MasterMetricManager::instance().get_global_mem_used_ratio();

        // 触发条件:
        // 1. 使用率超过高水位线
        // 2. need_eviction_ 标志被 PutStart 设置 (空间不足时)
        if (used_ratio > eviction_high_watermark_ratio_ ||
            (need_eviction_ && eviction_ratio_ > 0.0)) {

            double evict_ratio_target = std::max(
                eviction_ratio_,
                used_ratio - eviction_high_watermark_ratio_ + eviction_ratio_);
            double evict_ratio_lowerbound = ...;

            BatchEvict(evict_ratio_target, evict_ratio_lowerbound);
        }

        std::this_thread::sleep_for(
            std::chrono::milliseconds(kEvictionThreadSleepMs));
    }
}

need_eviction_ = true;
return tl::make_unexpected(ErrorCode::NO_AVAILABLE_HANDLE);

// L1205-1290
for (size_t i = 0; i < metadata_shards_.size(); i++) {
    auto& shard = metadata_shards_[(start_idx + i) % metadata_shards_.size()];
    MutexLocker lock(&shard.mutex);

    // 先清理过期的 processing keys
    DiscardExpiredProcessingKeys(shard, now);

    for (auto it = shard.metadata.begin(); it != shard.metadata.end(); it++) {
        // 跳过有 Lease 或未完成的对象
        if (!it->second.IsLeaseExpired(now) ||
            it->second.HasDiffRepStatus(ReplicaStatus::COMPLETE, ReplicaType::MEMORY)) {
            continue;
        }
        if (!it->second.IsSoftPinned(now)) {
            candidates.push_back(it->second.lease_timeout);
        }
    }

    // 使用 nth_element 找到第 N 小的 lease_timeout 作为阈值
    std::nth_element(candidates.begin(),
                     candidates.begin() + (evict_num - 1),
                     candidates.end());
    auto target_timeout = candidates[evict_num - 1];

    // 淘汰 lease_timeout <= target_timeout 的对象
    while (it != shard.metadata.end()) {
        if (it->second.lease_timeout <= target_timeout) {
            total_freed_size += it->second.size * it->second.GetMemReplicaCount();
            it->second.EraseReplica(ReplicaType::MEMORY);
            // ...
        }
    }
}

void MasterService::DiscardExpiredProcessingKeys(
    MetadataShard& shard, const std::chrono::steady_clock::time_point& now) {
    for (auto key_it = shard.processing_keys.begin(); ...) {
        auto& metadata = it->second;

        // PutStart 超时检测
        const auto ttl = metadata.put_start_time + put_start_release_timeout_sec_;
        if (ttl < now) {
            auto replicas = metadata.DiscardProcessingReplicas();
            // 放入 discarded_replicas_ 列表等待释放
            discarded_replicas_.emplace_back(std::move(replicas), ttl);
            shard.metadata.erase(it);
        }
    }
}

if (evicted_count > 0 || released_discarded_cnt > 0) {
    need_eviction_ = false;
    MasterMetricManager::instance().inc_eviction_success(evicted_count, total_freed_size);
} else {
    if (object_count == 0) {
        need_eviction_ = false;
    }
    MasterMetricManager::instance().inc_eviction_fail();
}