// 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());

}

}

void record_warmup_ed_finish_segment(const std::string& job_id_str, int64_t segment_size) {

g_warmup_ed_finish_segment_num.put({job_id_str}, 1);

g_warmup_ed_finish_segment_size.put({job_id_str}, segment_size);

update_warmup_ed_last_finish_ts(job_id_str);

}

void record_warmup_ed_finish_index(const std::string& job_id_str, int64_t idx_size) {

g_warmup_ed_finish_index_num.put({job_id_str}, 1);

g_warmup_ed_finish_index_size.put({job_id_str}, idx_size);

update_warmup_ed_last_finish_ts(job_id_str);

}

void record_warmup_ed_fail_segment(const std::string& job_id_str, int64_t segment_size) {

g_warmup_ed_fail_segment_num.put({job_id_str}, 1);

g_warmup_ed_fail_segment_size.put({job_id_str}, segment_size);

}

void record_warmup_ed_fail_index(const std::string& job_id_str, int64_t idx_size) {

g_warmup_ed_fail_index_num.put({job_id_str}, 1);

g_warmup_ed_fail_index_size.put({job_id_str}, idx_size);

}

void record_warmup_ed_skipped_rowset_as_finished(RowsetMeta& rs_meta,

const std::string& job_id_str) {

auto schema_ptr = rs_meta.tablet_schema();

bool has_inverted_index = schema_ptr->has_inverted_index() || schema_ptr->has_ann_index();

auto idx_version = schema_ptr->get_inverted_index_storage_format();

for (int64_t segment_id = 0; segment_id < rs_meta.num_segments(); segment_id++) {

record_warmup_ed_finish_segment(job_id_str, rs_meta.segment_file_size(segment_id));

if (!has_inverted_index) {

continue;

}

auto&& inverted_index_info = rs_meta.inverted_index_file_info(segment_id);

if (idx_version == InvertedIndexStorageFormatPB::V1) {

std::unordered_map<int64_t, int64_t> index_size_map;

for (const auto& info : inverted_index_info.index_info()) {

if (info.index_file_size() != -1) {

index_size_map[info.index_id()] = info.index_file_size();

} else {