// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License.  You may obtain a copy of the License at

//

//   http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied.  See the License for the

// specific language governing permissions and limitations

// under the License.

#include "cloud/cloud_internal_service.h"

#include <brpc/controller.h>

#include <bthread/countdown_event.h>

#include <butil/iobuf.h>

#include <fmt/format.h>

#include <algorithm>

#include <chrono>

#include <limits>

#include <list>

#include <memory>

#include <optional>

#include <thread>

#include <unordered_map>

#include <vector>

#include "cloud/cloud_storage_engine.h"

#include "cloud/cloud_tablet.h"

#include "cloud/cloud_tablet_mgr.h"

#include "cloud/cloud_warm_up_manager.h"

#include "cloud/cloud_warmup_metrics.h"

#include "cloud/config.h"

#include "io/cache/block_file_cache.h"

#include "io/cache/block_file_cache_downloader.h"

#include "io/cache/block_file_cache_factory.h"

#include "io/fs/path.h"

#include "runtime/thread_context.h"

#include "runtime/workload_management/io_throttle.h"

#include "storage/storage_policy.h"

#include "util/async_io.h"

#include "util/bvar_windowed_adder.h"

#include "util/debug_points.h"

namespace doris {

#include "common/compile_check_avoid_begin.h"

bvar::Adder<uint64_t> g_file_cache_get_by_peer_num("file_cache_get_by_peer_num");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_blocks_num("file_cache_get_by_peer_blocks_num");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_success_num("file_cache_get_by_peer_success_num");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_failed_num("file_cache_get_by_peer_failed_num");

bvar::LatencyRecorder g_file_cache_get_by_peer_server_latency(

        "file_cache_get_by_peer_server_latency");

bvar::LatencyRecorder g_file_cache_get_by_peer_read_cache_file_latency(

        "file_cache_get_by_peer_read_cache_file_latency");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_offer_failed_num(

        "file_cache_get_by_peer_offer_failed_num");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_queue_timeout_num(

        "file_cache_get_by_peer_queue_timeout_num");

bvar::LatencyRecorder g_file_cache_get_by_peer_queue_wait_latency(

        "file_cache_get_by_peer_queue_wait_latency");

bvar::LatencyRecorder g_file_cache_get_by_peer_handle_cache_block_req_latency(

        "file_cache_get_by_peer_handle_cache_block_req_latency");

bvar::LatencyRecorder g_file_cache_get_by_peer_get_cache_latency(

        "file_cache_get_by_peer_get_cache_latency");

bvar::LatencyRecorder g_file_cache_get_by_peer_get_or_set_latency(

        "file_cache_get_by_peer_get_or_set_latency");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_get_or_set_calls(

        "file_cache_get_by_peer_get_or_set_calls");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_get_or_set_blocks_total(

        "file_cache_get_by_peer_get_or_set_blocks_total");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_request_blocks_total(

        "file_cache_get_by_peer_request_blocks_total");

bvar::LatencyRecorder g_file_cache_get_by_peer_request_blocks_per_rpc(

        "file_cache_get_by_peer_request_blocks_per_rpc");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_response_blocks_total(

        "file_cache_get_by_peer_response_blocks_total");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_response_bytes_total(

        "file_cache_get_by_peer_response_bytes_total");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_not_downloaded_block_num(

        "file_cache_get_by_peer_not_downloaded_block_num");

bvar::LatencyRecorder g_file_cache_get_by_peer_read_file_block_total_latency(

        "file_cache_get_by_peer_read_file_block_total_latency");

bvar::LatencyRecorder g_file_cache_get_by_peer_set_response_data_latency(

        "file_cache_get_by_peer_set_response_data_latency");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_attachment_response_num(

        "file_cache_get_by_peer_attachment_response_num");

bvar::Adder<uint64_t> g_file_cache_get_by_peer_pb_response_num(

        "file_cache_get_by_peer_pb_response_num");

bvar::Adder<int64_t> g_peer_server_fill_requested("peer_server_fill_requested");

bvar::Adder<int64_t> g_peer_server_fill_success("peer_server_fill_success");

bvar::Adder<int64_t> g_peer_server_fill_timeout("peer_server_fill_timeout");

bvar::Adder<int64_t> g_peer_server_fill_rejected("peer_server_fill_rejected");

bvar::LatencyRecorder g_peer_server_fill_latency("peer_server_fill_latency");

bvar::LatencyRecorder g_cloud_internal_service_get_file_cache_meta_by_tablet_id_latency(

        "cloud_internal_service_get_file_cache_meta_by_tablet_id_latency");

// Concurrency guard for server-side S3 pull-through fills.

static std::atomic<int32_t> g_active_server_fills {0};

bvar::PassiveStatus<int32_t> g_peer_active_fills(

        "peer_active_fills",

        [](void*) { return g_active_server_fills.load(std::memory_order_relaxed); }, nullptr);

CloudInternalServiceImpl::CloudInternalServiceImpl(CloudStorageEngine& engine, ExecEnv* exec_env)

        : PInternalService(exec_env), _engine(engine) {}

CloudInternalServiceImpl::~CloudInternalServiceImpl() = default;

void CloudInternalServiceImpl::alter_vault_sync(google::protobuf::RpcController* controller,

                                                const doris::PAlterVaultSyncRequest* request,

                                                PAlterVaultSyncResponse* response,

                                                google::protobuf::Closure* done) {

    LOG(INFO) << "alter be to sync vault info from Meta Service";

    // If the vaults containing hdfs vault then it would try to create hdfs connection using jni

    // which would acuiqre one thread local jniEnv. But bthread context can't guarantee that the brpc

    // worker thread wouldn't do bthread switch between worker threads.

    bool ret = _heavy_work_pool.try_offer([this, done]() {

        brpc::ClosureGuard closure_guard(done);

        _engine.sync_storage_vault();

    });

    if (!ret) {

        brpc::ClosureGuard closure_guard(done);

        LOG(WARNING) << "fail to offer alter_vault_sync request to the work pool, pool="

                     << _heavy_work_pool.get_info();

    }

}

FileCacheType cache_type_to_pb(io::FileCacheType type) {

    switch (type) {

    case io::FileCacheType::TTL:

        return FileCacheType::TTL;

    case io::FileCacheType::INDEX:

        return FileCacheType::INDEX;

    case io::FileCacheType::NORMAL:

        return FileCacheType::NORMAL;

    default:

        DCHECK(false);

    }

    return FileCacheType::NORMAL;

}

static int64_t current_unix_time_us() {

    return std::chrono::duration_cast<std::chrono::microseconds>(

                   std::chrono::system_clock::now().time_since_epoch())

            .count();

}

static std::optional<int64_t> warm_up_rowset_cross_host_latency_us(int64_t start_unix_ts_us,

                                                                   int64_t end_unix_ts_us) {

    // The start timestamp is generated by the caller BE. Mixed-version callers may omit it, and

    // system clocks across BEs are not guaranteed to be ordered.

    if (start_unix_ts_us <= 0 || end_unix_ts_us < start_unix_ts_us) {

        return std::nullopt;

    }

    return end_unix_ts_us - start_unix_ts_us;

}

static void add_file_cache_block_meta_to_response(

        PGetFileCacheMetaResponse* resp, int64_t tablet_id, const std::string& rowset_id,

        int32_t segment_id, const std::string& file_name,

        const std::tuple<int64_t, int64_t, io::FileCacheType, int64_t>& tuple,

        const RowsetSharedPtr& rowset, bool is_index) {

    FileCacheBlockMeta* meta = resp->add_file_cache_block_metas();

    meta->set_tablet_id(tablet_id);

    meta->set_rowset_id(rowset_id);

    meta->set_segment_id(segment_id);

    meta->set_file_name(file_name);

    if (!is_index) {

        // .dat

        meta->set_file_size(rowset->rowset_meta()->segment_file_size(segment_id));

        meta->set_file_type(doris::FileType::SEGMENT_FILE);

    } else {

        // .idx

        const auto& idx_file_info = rowset->rowset_meta()->inverted_index_file_info(segment_id);

        meta->set_file_size(idx_file_info.has_index_size() ? idx_file_info.index_size() : -1);

        meta->set_file_type(doris::FileType::INVERTED_INDEX_FILE);

    }

    meta->set_offset(std::get<0>(tuple));

    meta->set_size(std::get<1>(tuple));

    meta->set_cache_type(cache_type_to_pb(std::get<2>(tuple)));

    meta->set_expiration_time(std::get<3>(tuple));

}

static void process_segment_file_cache_meta(PGetFileCacheMetaResponse* resp,

                                            const RowsetSharedPtr& rowset, int64_t tablet_id,

                                            const std::string& rowset_id, int32_t segment_id,

                                            bool is_index) {

    const char* extension = is_index ? ".idx" : ".dat";

    std::string file_name = fmt::format("{}_{}{}", rowset_id, segment_id, extension);

    auto cache_key = io::BlockFileCache::hash(file_name);

    auto* cache = io::FileCacheFactory::instance()->get_by_path(cache_key);

    if (!cache) return;

    auto segments_meta = cache->get_hot_blocks_meta(cache_key);

    for (const auto& tuple : segments_meta) {

        add_file_cache_block_meta_to_response(resp, tablet_id, rowset_id, segment_id, file_name,

                                              tuple, rowset, is_index);

    }

}

void CloudInternalServiceImpl::get_file_cache_meta_by_tablet_id(

        google::protobuf::RpcController* controller [[maybe_unused]],

        const PGetFileCacheMetaRequest* request, PGetFileCacheMetaResponse* response,

        google::protobuf::Closure* done) {

    brpc::ClosureGuard closure_guard(done);

    if (!config::enable_file_cache) {

        LOG_WARNING("try to access tablet file cache meta, but file cache not enabled");

        return;

    }

    auto begin_ts = std::chrono::duration_cast<std::chrono::microseconds>(

                            std::chrono::steady_clock::now().time_since_epoch())

                            .count();

    std::ostringstream tablet_ids_stream;

    int count = 0;

    for (const auto& tablet_id : request->tablet_ids()) {

        tablet_ids_stream << tablet_id << ", ";

        count++;

        if (count >= 10) {

            break;

        }

    }

    LOG(INFO) << "warm up get meta from this be, tablets num=" << request->tablet_ids().size()

              << ", first 10 tablet_ids=[ " << tablet_ids_stream.str() << " ]";

    for (const auto& tablet_id : request->tablet_ids()) {

        auto res = _engine.tablet_mgr().get_tablet(tablet_id);

        if (!res.has_value()) {

            LOG(ERROR) << "failed to get tablet: " << tablet_id

                       << " err msg: " << res.error().msg();

            continue;

        }

        CloudTabletSPtr tablet = std::move(res.value());

        auto st = tablet->sync_rowsets();

        if (!st) {

            // just log failed, try it best

            LOG(WARNING) << "failed to sync rowsets: " << tablet_id

                         << " err msg: " << st.to_string();

        }

        auto rowsets = tablet->get_snapshot_rowset();

        for (const RowsetSharedPtr& rowset : rowsets) {

            std::string rowset_id = rowset->rowset_id().to_string();

            for (int32_t segment_id = 0; segment_id < rowset->num_segments(); ++segment_id) {

                process_segment_file_cache_meta(response, rowset, tablet_id, rowset_id, segment_id,

                                                false);

                process_segment_file_cache_meta(response, rowset, tablet_id, rowset_id, segment_id,

                                                true);

            }

        }

    }

    auto end_ts = std::chrono::duration_cast<std::chrono::microseconds>(

                          std::chrono::steady_clock::now().time_since_epoch())

                          .count();

    g_cloud_internal_service_get_file_cache_meta_by_tablet_id_latency << (end_ts - begin_ts);

    LOG(INFO) << "get file cache meta by tablet ids = [ " << tablet_ids_stream.str() << " ] took "

              << end_ts - begin_ts << " us";

    VLOG_DEBUG << "get file cache meta by tablet id request=" << request->DebugString()

               << ", response=" << response->DebugString();

}

namespace {

// Helper functions for fetch_peer_data

inline int64_t elapsed_us(std::chrono::steady_clock::time_point start) {

    return std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() -

                                                                 start)

            .count();

}

int64_t get_rowset_meta_tablet_id_from_request(const PFetchPeerDataRequest* request) {

    return request->has_rowset_meta() && request->rowset_meta().has_tablet_id()

                   ? request->rowset_meta().tablet_id()

                   : -1;

}

std::string get_rowset_meta_resource_id_from_request(const PFetchPeerDataRequest* request) {

    if (request->has_rowset_meta() && request->rowset_meta().has_resource_id()) {

        return request->rowset_meta().resource_id();

    }

    return "";

}

std::string get_peer_cache_filename(std::string_view path) {

    return io::Path(std::string(path)).filename().native();

}

std::string format_peer_request_context(const PFetchPeerDataRequest* request,

                                        const io::UInt128Wrapper& hash, size_t file_size) {

    const std::string file_size_str =

            request->has_file_size() ? std::to_string(request->file_size()) : "unknown";

    return fmt::format(

            "type={}, path={}, cache_hash={}, request_fill={}, fill_tablet_id={}, "

            "fill_remote_path={}, fill_resource_id={}, file_size={}, resolved_file_size={}, "

            "cache_req_count={}, support_attachment={}",

            request->type(), request->path(), hash.to_string(),

            request->has_request_cache_fill() && request->request_cache_fill(),

            get_rowset_meta_tablet_id_from_request(request), request->path(),

            get_rowset_meta_resource_id_from_request(request), file_size_str,

            file_size == std::numeric_limits<size_t>::max() ? std::string("unknown")

                                                            : std::to_string(file_size),

            request->cache_req_size(),

            request->has_support_attachment() && request->support_attachment());

}

std::string format_peer_cache_block_context(const PFetchPeerDataRequest* request,

                                            const CacheBlockReqest& cb_req,

                                            const io::FileBlockSPtr& fb,

                                            const io::UInt128Wrapper& hash, size_t file_size,

                                            bool do_fill) {

    return fmt::format("{}, req_block=[offset={}, size={}], do_fill={}, block={}, cache_file={}",

                       format_peer_request_context(request, hash, file_size), cb_req.block_offset(),

                       cb_req.block_size(), do_fill, fb->get_info_for_log(), fb->get_cache_file());

}

std::string format_peer_fill_context(const io::FileBlockSPtr& fb, int64_t fill_tablet_id,

                                     const std::string& filename, const std::string& resource_id,

                                     const std::string& remote_path, int64_t file_size,

                                     int64_t offset, int64_t size, int32_t timeout_ms) {

    return fmt::format(

            "tablet_id={}, filename={}, resource_id={}, remote_path={}, file_size={}, "

            "request_range=[offset={}, size={}], timeout_ms={}, block={}, cache_file={}",

            fill_tablet_id, filename, resource_id.empty() ? "<unknown>" : resource_id,

            remote_path.empty() ? "<unknown>" : remote_path, file_size, offset, size, timeout_ms,

            fb->get_info_for_log(), fb->get_cache_file());

}

bool wait_for_file_block_state(const io::FileBlockSPtr& fb, int32_t timeout_ms) {

    const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms);

    while (true) {

        const auto state = fb->state();

        if (state == io::FileBlock::State::DOWNLOADED ||

            state == io::FileBlock::State::SKIP_CACHE || state == io::FileBlock::State::EMPTY) {

            return true;

        }

        if (std::chrono::steady_clock::now() >= deadline) {

            return false;

        }

        fb->wait();

    }

}

Status handle_peer_file_range_request(const std::string& path, PFetchPeerDataResponse* response) {

    // Legacy path: PEER_FILE_RANGE still returns payload via protobuf bytes.

    // Keep this for compatibility until range path is migrated to attachment mode.

    // Read specific range [file_offset, file_offset+file_size) across cached blocks

    auto datas =

            io::FileCacheFactory::instance()->get_cache_data_by_path(get_peer_cache_filename(path));

    for (auto& cb : datas) {

        *(response->add_datas()) = std::move(cb);

    }

    return Status::OK();

}

void set_error_response(PFetchPeerDataResponse* response, const std::string& error_msg) {

    response->mutable_status()->add_error_msgs(error_msg);

    response->mutable_status()->set_status_code(TStatusCode::INTERNAL_ERROR);

}

void set_too_many_tasks_response(PFetchPeerDataResponse* response, const std::string& error_msg) {

    response->mutable_status()->add_error_msgs(error_msg);

    response->mutable_status()->set_status_code(TStatusCode::TOO_MANY_TASKS);

}

bool try_reject_if_queue_timed_out(std::chrono::steady_clock::time_point enqueue_ts,

                                   PFetchPeerDataResponse* response) {

    auto wait_us = elapsed_us(enqueue_ts);

    g_file_cache_get_by_peer_queue_wait_latency << wait_us;

    auto wait_ms = wait_us / 1000;

    if (wait_ms <= config::peer_fetch_queue_timeout_ms) {

        return false;

    }

    const std::string msg = fmt::format("fetch peer data queue timeout, wait_ms={}, timeout_ms={}",

                                        wait_ms, config::peer_fetch_queue_timeout_ms);

    g_file_cache_get_by_peer_queue_timeout_num << 1;

    set_too_many_tasks_response(response, msg);

    return true;

}

Status read_file_block(const std::shared_ptr<io::FileBlock>& file_block, size_t file_size,

                       doris::CacheBlockPB* output, butil::IOBuf* response_attachment) {

    auto total_start = std::chrono::steady_clock::now();

    int64_t set_data_us = 0;

    Defer report {[&]() {

        g_file_cache_get_by_peer_read_file_block_total_latency << elapsed_us(total_start);

        if (set_data_us > 0) {

            g_file_cache_get_by_peer_set_response_data_latency << set_data_us;

        }

    }};

    // ATTN: calculate the rightmost boundary value of the block, due to inaccurate current block meta information.

    // see CachedRemoteFileReader::read_at_impl for more details.

    // Ensure file_size >= file_block->offset() to avoid underflow

    if (file_size < file_block->offset()) {

        LOG(WARNING) << "file_size (" << file_size << ") < file_block->offset("

                     << file_block->offset() << ")";

        return Status::InternalError<false>("file_size less than block offset");

    }

    size_t read_size = std::min(static_cast<size_t>(file_size - file_block->offset()),

                                file_block->range().size());

    output->set_block_offset(static_cast<int64_t>(file_block->offset()));

    output->set_block_size(static_cast<int64_t>(read_size));

    if (read_size == 0) {

        return Status::OK();

    }

    Status read_st = Status::OK();

    // Attachment payload mode: protobuf carries metadata only, payload goes to attachment.

    // This allows FS cache to use a file-descriptor->IOBuf path directly.

    if (response_attachment != nullptr) {

        size_t bytes_read = 0;

        auto begin_read_file_ts = std::chrono::duration_cast<std::chrono::microseconds>(

                                          std::chrono::steady_clock::now().time_since_epoch())

                                          .count();

        SCOPED_ATTACH_TASK(ExecEnv::GetInstance()->s3_file_buffer_tracker());

        read_st = file_block->read_to_iobuf(response_attachment, /*read_offset=*/0, read_size,

                                            &bytes_read);

        auto end_read_file_ts = std::chrono::duration_cast<std::chrono::microseconds>(

                                        std::chrono::steady_clock::now().time_since_epoch())

                                        .count();

        g_file_cache_get_by_peer_read_cache_file_latency << (end_read_file_ts - begin_read_file_ts);

        if (read_st.ok()) {

            if (bytes_read != read_size) {

                return Status::InternalError<false>(

                        "peer cache read short data, expected={}, actual={}", read_size,

                        bytes_read);

            }

            g_file_cache_get_by_peer_response_bytes_total << bytes_read;

            return Status::OK();

        }

    } else {

        std::string data;

        data.resize(read_size);

        auto begin_read_file_ts = std::chrono::duration_cast<std::chrono::microseconds>(

                                          std::chrono::steady_clock::now().time_since_epoch())

                                          .count();

        SCOPED_ATTACH_TASK(ExecEnv::GetInstance()->s3_file_buffer_tracker());

        Slice slice(data.data(), data.size());

        read_st = file_block->read(slice, /*read_offset=*/0);

        auto end_read_file_ts = std::chrono::duration_cast<std::chrono::microseconds>(

                                        std::chrono::steady_clock::now().time_since_epoch())

                                        .count();

        g_file_cache_get_by_peer_read_cache_file_latency << (end_read_file_ts - begin_read_file_ts);

        if (read_st.ok()) {

            auto set_data_start = std::chrono::steady_clock::now();

            output->set_data(std::move(data));

            set_data_us = elapsed_us(set_data_start);

            g_file_cache_get_by_peer_response_bytes_total << read_size;

            return Status::OK();

        }

    }

    g_file_cache_get_by_peer_failed_num << 1;

    LOG(WARNING) << "read cache block failed, file_size=" << file_size

                 << ", block=" << file_block->get_info_for_log()

                 << ", cache_file=" << file_block->get_cache_file() << ", err=" << read_st;

    return read_st;

}

// Trigger S3 -> local cache fill for the given file block.

// Returns OK when the block is DOWNLOADED after the fill.

// Returns TOO_MANY_TASKS when the fill slot is exhausted (server healthy but overloaded):

//   client should not rotate or evict, just fall back to S3 and retry same candidate later.

// Returns NOT_FOUND for soft misses (tablet not found, fill incomplete, timeout):

//   client should rotate the candidate to try a different CG next time.

// The peer uses request.path as the full remote path. tablet_id/filename are kept for logging.

Status trigger_peer_server_fill(io::FileBlockSPtr& fb, int64_t fill_tablet_id,

                                const std::string& filename, const std::string& resource_id,

                                const std::string& remote_path, int64_t file_size, int64_t offset,

                                int64_t size, int32_t timeout_ms) {

    g_peer_server_fill_requested << 1;

    // Concurrency guard: atomically reserve a fill slot.

    // Excess requests are rejected so the client falls back to its own S3 read.

    // Return NOT_FOUND so the client rotates the candidate instead of evicting it.

    if (g_active_server_fills.fetch_add(1, std::memory_order_relaxed) >=

        config::max_concurrent_peer_server_fills) {

        g_active_server_fills.fetch_sub(1, std::memory_order_relaxed);

        g_peer_server_fill_rejected << 1;

        VLOG_DEBUG << "trigger_peer_server_fill: rejected (concurrency limit "

                   << config::max_concurrent_peer_server_fills << "), tablet_id=" << fill_tablet_id;

        // TOO_MANY_TASKS: server is healthy but overloaded. Client must not rotate or evict;

        // just fall back to S3 for this request and retry the same candidate next time.

        return Status::Error<ErrorCode::TOO_MANY_TASKS, false>("fill slot exhausted");

    }

    // RAII decrement: runs on every return path below.

    Defer fill_guard {[]() { g_active_server_fills.fetch_sub(1, std::memory_order_relaxed); }};

    if (remote_path.empty() || resource_id.empty()) {

        const std::string ctx =

                format_peer_fill_context(fb, fill_tablet_id, filename, resource_id, remote_path,

                                         file_size, offset, size, timeout_ms);

        LOG(WARNING) << "trigger_peer_server_fill: missing remote_path or resource_id, " << ctx;

        g_peer_server_fill_rejected << 1;

        return Status::NotFound<false>("fill: missing remote_path or resource_id, {}", ctx);

    }

    auto storage_resource = doris::get_storage_resource(resource_id);

    if (!storage_resource.has_value()) {

        const std::string ctx =

                format_peer_fill_context(fb, fill_tablet_id, filename, resource_id, remote_path,

                                         file_size, offset, size, timeout_ms);

        LOG(WARNING) << "trigger_peer_server_fill: storage resource not found, " << ctx;

        g_peer_server_fill_rejected << 1;

        return Status::NotFound<false>("fill: storage resource not found, {}", ctx);

    }

    auto fs = storage_resource->first.fs;

    const auto initial_state = fb->state();

    if (initial_state == io::FileBlock::State::DOWNLOADING) {

        // Another thread already owns the block downloader. Wait up to the request timeout instead

        // of the shorter per-wait timeout in FileBlock::wait().

        [[maybe_unused]] const bool completed = wait_for_file_block_state(fb, timeout_ms);

        const std::string ctx =

                format_peer_fill_context(fb, fill_tablet_id, filename, resource_id, remote_path,

                                         file_size, offset, size, timeout_ms);

        return fb->state() == io::FileBlock::State::DOWNLOADED

                       ? Status::OK()

                       : Status::NotFound<false>("fill: concurrent download incomplete, {}", ctx);

    }

    if (initial_state != io::FileBlock::State::EMPTY) {

        const std::string ctx =

                format_peer_fill_context(fb, fill_tablet_id, filename, resource_id, remote_path,

                                         file_size, offset, size, timeout_ms);

        return initial_state == io::FileBlock::State::DOWNLOADED

                       ? Status::OK()

                       : Status::NotFound<false>("fill: unexpected initial block state, {}", ctx);

    }

    auto fill_start = std::chrono::steady_clock::now();

    auto fill_done = std::make_shared<bthread::CountdownEvent>(1);

    auto fill_status = std::make_shared<Status>(Status::OK());

    io::DownloadFileMeta download_meta {

            .path = remote_path,

            .file_size = file_size,

            .offset = offset,

            .download_size = size,

            .file_system = fs,

            .ctx = {.is_dryrun = config::enable_reader_dryrun_when_download_file_cache,

                    // Pull-through fill must go straight to remote storage. If this download

                    // re-enters peer race, the original block can remain DOWNLOADING for the

                    // duration of nested peer retries and timeouts.

                    .is_warmup = false,

                    .bypass_peer_read = true},

            .download_done =

                    [fill_done, fill_status](Status st) {

                        *fill_status = std::move(st);

                        fill_done->signal();

                    },

            .tablet_id = fill_tablet_id,

    };

    io::DownloadTask task(std::move(download_meta));

    ExecEnv::GetInstance()

            ->storage_engine()

            .to_cloud()

            .file_cache_block_downloader()

            .submit_download_task(std::move(task));

    const timespec due_time = butil::milliseconds_from_now(timeout_ms);

    const bool timed_out = fill_done->timed_wait(due_time) != 0;

    int64_t fill_ms = std::chrono::duration_cast<std::chrono::milliseconds>(

                              std::chrono::steady_clock::now() - fill_start)

                              .count();

    g_peer_server_fill_latency << fill_ms * 1000; // LatencyRecorder takes microseconds

    if (!timed_out && fill_status->ok() && fb->state() == io::FileBlock::State::DOWNLOADING) {

        const int32_t settle_timeout_ms =

                std::max<int32_t>(1, timeout_ms - static_cast<int32_t>(fill_ms));

        [[maybe_unused]] const bool settled = wait_for_file_block_state(fb, settle_timeout_ms);

    }

    auto final_state = fb->state();

    if (final_state == io::FileBlock::State::DOWNLOADED) {

        g_peer_server_fill_success << 1;

        return Status::OK();

    }

    if (timed_out) {

        LOG(WARNING) << "trigger_peer_server_fill: fill timeout, elapsed_ms=" << fill_ms << ", "

                     << format_peer_fill_context(fb, fill_tablet_id, filename, resource_id,

                                                 remote_path, file_size, offset, size, timeout_ms);

        g_peer_server_fill_timeout << 1;

    } else if (!fill_status->ok()) {

        LOG(WARNING) << "trigger_peer_server_fill: fill failed, elapsed_ms=" << fill_ms

                     << ", status=" << fill_status->to_string() << ", "

                     << format_peer_fill_context(fb, fill_tablet_id, filename, resource_id,

                                                 remote_path, file_size, offset, size, timeout_ms);

    }

    // Any non-DOWNLOADED outcome is a soft failure: the server is otherwise healthy so the

    // client should rotate the candidate rather than evict it.

    return Status::NotFound<false>(

            "fill: block not downloaded, {}",

            format_peer_fill_context(fb, fill_tablet_id, filename, resource_id, remote_path,

                                     file_size, offset, size, timeout_ms));

}

Status handle_peer_file_cache_block_request(const PFetchPeerDataRequest* request,

                                            PFetchPeerDataResponse* response,

                                            brpc::Controller* cntl) {

    auto handle_start = std::chrono::steady_clock::now();

    const uint64_t request_blocks = request->cache_req_size();

    uint64_t response_blocks = 0;

    uint64_t get_or_set_calls = 0;

    uint64_t get_or_set_blocks = 0;

    int64_t get_cache_us = 0;

    Defer report {[&]() {

        g_file_cache_get_by_peer_handle_cache_block_req_latency << elapsed_us(handle_start);

        g_file_cache_get_by_peer_get_cache_latency << get_cache_us;

        g_file_cache_get_by_peer_get_or_set_calls << get_or_set_calls;

        g_file_cache_get_by_peer_get_or_set_blocks_total << get_or_set_blocks;

        g_file_cache_get_by_peer_request_blocks_total << request_blocks;

        g_file_cache_get_by_peer_request_blocks_per_rpc << request_blocks;

        g_file_cache_get_by_peer_response_blocks_total << response_blocks;

    }};

    const auto& path = request->path();

    const auto cache_key_path = get_peer_cache_filename(path);

    auto hash = io::BlockFileCache::hash(cache_key_path);

    auto get_cache_start = std::chrono::steady_clock::now();

    auto* cache = io::FileCacheFactory::instance()->get_by_path(hash);

    get_cache_us = elapsed_us(get_cache_start);

    if (cache == nullptr) {

        g_file_cache_get_by_peer_failed_num << 1;

        set_error_response(response, "can't get file cache instance");

        return Status::InternalError<false>("can't get file cache instance");

    }

    io::CacheContext ctx {};

    io::ReadStatistics local_stats;

    ctx.stats = &local_stats;

    const size_t file_size =

            request->has_file_size()

                    ? static_cast<size_t>(std::max<int64_t>(0, request->file_size()))

                    : std::numeric_limits<size_t>::max();

    // Enable attachment mode only when client advertises support.

    // This keeps mixed-version rolling upgrades safe.

    const bool use_attachment =

            cntl != nullptr && request->has_support_attachment() && request->support_attachment();

    response->set_data_in_attachment(use_attachment);

    if (use_attachment) {

        g_file_cache_get_by_peer_attachment_response_num << 1;

    } else {

        g_file_cache_get_by_peer_pb_response_num << 1;

    }

    const bool do_fill = request->has_request_cache_fill() && request->request_cache_fill() &&

                         config::enable_peer_server_cache_fill;

    for (const auto& cb_req : request->cache_req()) {

        size_t offset = static_cast<size_t>(std::max<int64_t>(0, cb_req.block_offset()));

        size_t size = static_cast<size_t>(std::max<int64_t>(0, cb_req.block_size()));

        if (offset >= file_size) {

            continue;

        }

        // Clip tail requests before get_or_set so peer reads do not synthesize EMPTY blocks past

        // EOF and then fail the whole RPC.

        size = std::min(size, file_size - offset);

        if (size == 0) {

            continue;

        }

        DBUG_EXECUTE_IF(

                "CloudInternalServiceImpl::handle_peer_file_cache_block_request_hold_before_get_or_"

                "set",

                {

                    int sleep_ms = dp->param<int>("sleep_ms", 300);

                    bthread_usleep(sleep_ms * 1000);

                });

        auto get_or_set_start = std::chrono::steady_clock::now();

        auto holder = cache->get_or_set(hash, offset, size, ctx);

        g_file_cache_get_by_peer_get_or_set_latency << elapsed_us(get_or_set_start);

        ++get_or_set_calls;

        get_or_set_blocks += holder.file_blocks.size();

        for (auto& fb : holder.file_blocks) {

            auto fb_state = fb->state();

            if (fb_state == io::FileBlock::State::DOWNLOADING) {

                if (do_fill) {

                    // Only peer fill requests should wait longer here. Plain peer-cache reads keep

                    // the short wait semantics so they can fail fast and let the client race S3.

                    [[maybe_unused]] const bool completed = wait_for_file_block_state(

                            fb, config::peer_server_cache_fill_timeout_ms);

                    fb_state = fb->state();

                } else {

                    // Wait for in-progress download to complete using the normal short timeout.

                    fb_state = fb->wait();

                }

            }

            if (fb_state == io::FileBlock::State::EMPTY) {

                if (!do_fill) {

                    const std::string msg =

                            fmt::format("cache block not downloaded, {}",

                                        format_peer_cache_block_context(request, cb_req, fb, hash,

                                                                        file_size, do_fill));

                    g_file_cache_get_by_peer_failed_num << 1;

                    g_file_cache_get_by_peer_not_downloaded_block_num << 1;

                    LOG(WARNING) << msg;

                    // Use NOT_FOUND so the client can distinguish "block not cached"

                    // from an actual RPC/server error.  On NOT_FOUND the client rotates

                    // the candidate to the end of its list (trying another CG next time)

                    // rather than incrementing the RPC-failure eviction counter.

                    response->mutable_status()->add_error_msgs(msg);

                    response->mutable_status()->set_status_code(TStatusCode::NOT_FOUND);

                    return Status::NotFound<false>(msg);

                }

                // Server-side fill: request.path already carries the full remote path.

                auto fill_st = trigger_peer_server_fill(

                        fb, get_rowset_meta_tablet_id_from_request(request), cache_key_path,

                        get_rowset_meta_resource_id_from_request(request), path,

                        request->has_file_size() ? request->file_size() : -1,

                        static_cast<int64_t>(fb->range().left),

                        static_cast<int64_t>(fb->range().size()),

                        config::peer_server_cache_fill_timeout_ms);

                if (!fill_st.ok()) {

                    g_file_cache_get_by_peer_failed_num << 1;

                    g_file_cache_get_by_peer_not_downloaded_block_num << 1;

                    if (fill_st.is<ErrorCode::TOO_MANY_TASKS>()) {

                        // Server slot exhausted: healthy but overloaded. Client must not rotate

                        // or evict — just fall back to S3 and retry same candidate next time.

                        response->mutable_status()->add_error_msgs(std::string(fill_st.msg()));

                        response->mutable_status()->set_status_code(TStatusCode::TOO_MANY_TASKS);

                    } else if (fill_st.is<ErrorCode::NOT_FOUND>()) {

                        // Soft miss (fill incomplete, timeout, unexpected state) — client rotates,

                        // not evicts.

                        response->mutable_status()->add_error_msgs(std::string(fill_st.msg()));

                        response->mutable_status()->set_status_code(TStatusCode::NOT_FOUND);

                    } else {

                        LOG(WARNING) << "cache block fill failed, status=" << fill_st << ", "

                                     << format_peer_cache_block_context(request, cb_req, fb, hash,

                                                                        file_size, do_fill);

                        set_error_response(response, "cache block not ready");

                    }

                    return fill_st;

                }

                fb_state = io::FileBlock::State::DOWNLOADED;

            }

            if (fb_state != io::FileBlock::State::DOWNLOADED) {

                // A concurrent download was in progress (DOWNLOADING at request time) but its

                // wait() returned a non-DOWNLOADED state (e.g., timed-out while still

                // DOWNLOADING, or some other non-EMPTY intermediate state).  The server is

                // healthy; the block just isn't available yet.  Return NOT_FOUND so the client

                // rotates the candidate instead of evicting it.

                const std::string msg =

                        fmt::format("cache block not ready after wait, {}",

                                    format_peer_cache_block_context(request, cb_req, fb, hash,

                                                                    file_size, do_fill));

                g_file_cache_get_by_peer_failed_num << 1;

                g_file_cache_get_by_peer_not_downloaded_block_num << 1;

                LOG(WARNING) << msg;

                response->mutable_status()->add_error_msgs(msg);

                response->mutable_status()->set_status_code(TStatusCode::NOT_FOUND);

                return Status::NotFound<false>(msg);

            }

            g_file_cache_get_by_peer_blocks_num << 1;

            doris::CacheBlockPB* out = response->add_datas();

            // In attachment mode, metadata order must match attachment append order because the

            // client consumes attachment payload sequentially using resp.datas() order.

            Status read_status = read_file_block(

                    fb, file_size, out, use_attachment ? &cntl->response_attachment() : nullptr);

            if (!read_status.ok()) {

                set_error_response(response, "read cache file error");

                return read_status;

            }

            ++response_blocks;

        }

    }

    return Status::OK();

}

} // namespace

#ifdef BE_TEST

Status test_handle_peer_file_cache_block_request(const PFetchPeerDataRequest* request,

                                                 PFetchPeerDataResponse* response,

                                                 brpc::Controller* cntl) {

    return handle_peer_file_cache_block_request(request, response, cntl);

}

bool test_try_reject_if_queue_timed_out(std::chrono::steady_clock::time_point enqueue_ts,

                                        PFetchPeerDataResponse* response) {

    return try_reject_if_queue_timed_out(enqueue_ts, response);

}

#endif

void CloudInternalServiceImpl::fetch_peer_data(google::protobuf::RpcController* controller,

                                               const PFetchPeerDataRequest* request,

                                               PFetchPeerDataResponse* response,

                                               google::protobuf::Closure* done) {

    auto enqueue_ts = std::chrono::steady_clock::now();

    // Lifetime: cntl is owned by brpc framework and valid until done->Run() is called.

    // The ClosureGuard inside the lambda ensures done->Run() happens after all cntl usage,

    // so capturing the raw pointer by value is safe.

    auto* cntl = static_cast<brpc::Controller*>(controller);

    bool ret = _peer_fetch_pool.try_offer([request, response, done, enqueue_ts, cntl]() {

        brpc::ClosureGuard closure_guard(done);

        g_file_cache_get_by_peer_num << 1;

        if (try_reject_if_queue_timed_out(enqueue_ts, response)) {

            return;

        }

        if (!config::enable_file_cache) {

            LOG_WARNING("try to access file cache data, but file cache not enabled");

            return;

        }

        auto begin_ts = std::chrono::duration_cast<std::chrono::microseconds>(

                                std::chrono::steady_clock::now().time_since_epoch())

                                .count();

        const auto type = request->type();

        const auto& path = request->path();

        response->mutable_status()->set_status_code(TStatusCode::OK);

        Status status = Status::OK();

        if (type == PFetchPeerDataRequest_Type_PEER_FILE_RANGE) {

            status = handle_peer_file_range_request(path, response);

        } else if (type == PFetchPeerDataRequest_Type_PEER_FILE_CACHE_BLOCK) {

            status = handle_peer_file_cache_block_request(request, response, cntl);

        }

        if (!status.ok()) {

            const std::string msg =

                    "fetch peer data failed: " + status.to_string() + ", " +

                    format_peer_request_context(

                            request, io::BlockFileCache::hash(get_peer_cache_filename(path)),

                            request->has_file_size() ? static_cast<size_t>(std::max<int64_t>(

                                                               0, request->file_size()))

                                                     : std::numeric_limits<size_t>::max());

            if (status.is<ErrorCode::NOT_FOUND>() || status.is<ErrorCode::TOO_MANY_TASKS>()) {

                VLOG_DEBUG << msg;

            } else {

                LOG(WARNING) << msg;

            }

            auto* resp_status = response->mutable_status();

            if (resp_status->status_code() == TStatusCode::OK) {

                set_error_response(response, status.to_string());

            } else if (resp_status->error_msgs().empty()) {

                resp_status->add_error_msgs(status.to_string());

            }

        }

        DBUG_EXECUTE_IF("CloudInternalServiceImpl::fetch_peer_data_slower", {

            int st_us = dp->param<int>("sleep", 1000);

            LOG_WARNING("CloudInternalServiceImpl::fetch_peer_data_slower").tag("sleep", st_us);

            bthread_usleep(st_us);

        });

        auto end_ts = std::chrono::duration_cast<std::chrono::microseconds>(

                              std::chrono::steady_clock::now().time_since_epoch())

                              .count();

        // Latency covers every completed callback (including failures) so the

        // server-side fail-fast paths still show up in the latency histogram.

        // success_num must only count actual OK results, otherwise dedup

        // TOO_MANY_TASKS / NOT_FOUND / handler errors all fall through here

        // and the success rate is meaningless. Use file_cache_get_by_peer_num

        // for the total completed-callback count.

        g_file_cache_get_by_peer_server_latency << (end_ts - begin_ts);

        if (status.ok()) {

            g_file_cache_get_by_peer_success_num << 1;

        }

        VLOG_DEBUG << "fetch cache request=" << request->DebugString()

                   << ", response=" << response->DebugString();

    });

    if (!ret) {

        g_file_cache_get_by_peer_offer_failed_num << 1;

        brpc::ClosureGuard closure_guard(done);

        const std::string msg = fmt::format(

                "fail to offer fetch peer data request to the peer fetch work pool, pool={}",

                _peer_fetch_pool.get_info());

        set_too_many_tasks_response(response, msg);

        LOG(WARNING) << msg;

    }

}

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_submitted_segment_num(

        "file_cache_event_driven_warm_up_submitted_segment_num");

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_finished_segment_num(

        "file_cache_event_driven_warm_up_finished_segment_num");

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_failed_segment_num(

        "file_cache_event_driven_warm_up_failed_segment_num");

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_submitted_segment_size(

        "file_cache_event_driven_warm_up_submitted_segment_size");

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_finished_segment_size(

        "file_cache_event_driven_warm_up_finished_segment_size");

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_failed_segment_size(

        "file_cache_event_driven_warm_up_failed_segment_size");

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_submitted_index_num(

        "file_cache_event_driven_warm_up_submitted_index_num");

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_finished_index_num(

        "file_cache_event_driven_warm_up_finished_index_num");

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_failed_index_num(

        "file_cache_event_driven_warm_up_failed_index_num");

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_submitted_index_size(

        "file_cache_event_driven_warm_up_submitted_index_size");

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_finished_index_size(

        "file_cache_event_driven_warm_up_finished_index_size");

bvar::Adder<uint64_t> g_file_cache_event_driven_warm_up_failed_index_size(

        "file_cache_event_driven_warm_up_failed_index_size");

bvar::Status<int64_t> g_file_cache_warm_up_rowset_last_handle_unix_ts(

        "file_cache_warm_up_rowset_last_handle_unix_ts", 0);

bvar::Status<int64_t> g_file_cache_warm_up_rowset_last_finish_unix_ts(

        "file_cache_warm_up_rowset_last_finish_unix_ts", 0);

bvar::LatencyRecorder g_file_cache_warm_up_rowset_latency("file_cache_warm_up_rowset_latency");

bvar::LatencyRecorder g_file_cache_warm_up_rowset_request_to_handle_latency(

        "file_cache_warm_up_rowset_request_to_handle_latency");

bvar::LatencyRecorder g_file_cache_warm_up_rowset_handle_to_finish_latency(

        "file_cache_warm_up_rowset_handle_to_finish_latency");

bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_slow_count(

        "file_cache_warm_up_rowset_slow_count");

bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_request_to_handle_slow_count(

        "file_cache_warm_up_rowset_request_to_handle_slow_count");

bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_handle_to_finish_slow_count(

        "file_cache_warm_up_rowset_handle_to_finish_slow_count");

bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_wait_for_compaction_num(

        "file_cache_warm_up_rowset_wait_for_compaction_num");

bvar::Adder<uint64_t> g_file_cache_warm_up_rowset_wait_for_compaction_timeout_num(

        "file_cache_warm_up_rowset_wait_for_compaction_timeout_num");

// Per-job windowed metrics for target BE

// bvar::Window enforces MAX_SECONDS_LIMIT = 3600, so the longest window is 1h.

static constexpr int WINDOW_5M = 300;

static constexpr int WINDOW_30M = 1800;

static constexpr int WINDOW_1H = 3600;

MBvarWindowedAdder g_warmup_ed_finish_segment_num("warmup_ed_finish_segment_num", {"job_id"},

                                                  {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);

MBvarWindowedAdder g_warmup_ed_finish_segment_size("warmup_ed_finish_segment_size", {"job_id"},

                                                   {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);

MBvarWindowedAdder g_warmup_ed_finish_index_num("warmup_ed_finish_index_num", {"job_id"},

                                                {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);

MBvarWindowedAdder g_warmup_ed_finish_index_size("warmup_ed_finish_index_size", {"job_id"},

                                                 {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);

MBvarWindowedAdder g_warmup_ed_fail_segment_num("warmup_ed_fail_segment_num", {"job_id"},

                                                {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);

MBvarWindowedAdder g_warmup_ed_fail_segment_size("warmup_ed_fail_segment_size", {"job_id"},

                                                 {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);

MBvarWindowedAdder g_warmup_ed_fail_index_num("warmup_ed_fail_index_num", {"job_id"},

                                              {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);

MBvarWindowedAdder g_warmup_ed_fail_index_size("warmup_ed_fail_index_size", {"job_id"},

                                               {WINDOW_5M, WINDOW_30M, WINDOW_1H}, false);

bvar::MultiDimension<bvar::Status<int64_t>> g_warmup_ed_last_finish_ts({"job_id"});

void update_warmup_ed_last_finish_ts(const std::string& job_id_str) {

    auto* finish_ts = g_warmup_ed_last_finish_ts.get_stats(std::list<std::string> {job_id_str});

    if (finish_ts) {

        finish_ts->set_value(std::chrono::duration_cast<std::chrono::milliseconds>(

                                     std::chrono::system_clock::now().time_since_epoch())

                                     .count());

    }

}