[improvement](be) Skip stale S3 race reads and serve peer reads without get_or_set (#69620)

Problem Summary:

1. Peer/S3 race issues S3 GETs that are already useless.
launch_s3_race() rechecks the race only after the hedge delay. Once the
S3 leg is submitted to PeerRaceS3ThreadPool, the peer can win while the
task is still queued. The task then allocates a span-sized buffer, sends
the S3 GET, reads the whole span and only afterwards sees winner == 0
and drops the result. This wastes S3 requests, network bandwidth and
pool threads, and makes later S3 tasks queue longer when the pool is
small.

Fix: do_s3_read() takes race->mtx before allocating the buffer or
counting the read. If the peer has already won, it marks the S3 leg as
skipped, notifies waiters and returns. A new bvar
peer_race_s3_skipped_after_queue counts these skips, which separates
them from S3 reads that were already in flight when the peer won. A new
sync point s3_task_start is placed before the check; the existing
s3_before_read stays after it, so existing tests that hold S3 there keep
their behavior.

2. Peer cache server uses get_or_set() for plain peer reads.
handle_peer_file_cache_block_request() called
BlockFileCache::get_or_set() for every requested block. For a plain read
(no pull-through fill) a miss inserted an EMPTY cell that was removed
again when the holder was released, and the server could take downloader
ownership of a block it never downloads.

Fix: plain peer reads use the read-only
BlockFileCache::get_downloaded_blocks_if_fully_covered(). It walks the
existing cells, returns them only when DOWNLOADED cells cover the whole
requested range, and refreshes their LRU position under the same cache
lock. It never creates cells or takes downloader ownership, and it
accepts any cached layout. If the range is not fully covered, the server
returns NOT_FOUND so the client rotates the candidate. Pull-through fill
requests still use get_or_set(), because the fill download lands in the
EMPTY cell it creates and the handler reads that cell back. The
DOWNLOADING/EMPTY/not-ready handling and the block read are shared by
both paths through serve_peer_cache_block().

BlockFileCache::probe() is not used here: it requires cached cells to be
aligned to this cache's block size and aborts on a DORIS_CHECK
otherwise. A pull-through fill from a client with a smaller cache block
size (or a cache directory written with an older block size) leaves
short cells inside one block, and probe() on such a layout terminates
the BE.

With enable_file_cache_async_touch_on_get_or_set=true, use_cell() skips
the LRU move and get_or_set() queues downloaded hits through
add_need_update_lru_block() instead.
get_downloaded_blocks_if_fully_covered() did not queue them, so blocks
served only through it never moved in LRU and were evicted ahead of
colder blocks. It now queues the covered blocks after releasing the
cache lock, as get_or_set() does. This also applies to the existing
remote-only-on-miss read path that uses the same lookup.

The plain-read lookup adds no metric: it is one in-memory cell walk
under the cache lock, and its cost is already part of
file_cache_get_by_peer_handle_cache_block_req_latency. The existing
get_or_set bvars now only count fill requests. The only new bvar in this
PR is peer_race_s3_skipped_after_queue (one Adder), which separates S3
GETs avoided because the peer won while the task was queued from S3
reads that were already in flight.

be/src/cloud/cloud_internal_service.cpp

@@ -446,6 +446,17 @@ void set_too_many_tasks_response(PFetchPeerDataResponse* response, const std::st
     response->mutable_status()->set_status_code(TStatusCode::TOO_MANY_TASKS);
 }
 
+// NOT_FOUND means this peer cannot serve the block now; the client rotates the candidate instead
+// of counting an RPC failure.
+Status reject_peer_block_not_found(PFetchPeerDataResponse* response, const std::string& msg) {
+    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);
+}
+
 bool try_reject_if_queue_timed_out(std::chrono::steady_clock::time_point enqueue_ts,
                                    PFetchPeerDataResponse* response) {
     auto wait_us = elapsed_us(enqueue_ts);
@@ -680,6 +691,87 @@ Status trigger_peer_server_fill(io::FileBlockSPtr& fb, int64_t fill_tablet_id,
                                      file_size, offset, size, timeout_ms));
 }
 
+// Serve one cache block of a peer request: wait for a DOWNLOADING block, fill an EMPTY block when
+// `do_fill` is set, then append the block to `response`. The payload goes to
+// `response_attachment` when it is non-null, otherwise into the protobuf.
+Status serve_peer_cache_block(const PFetchPeerDataRequest* request, const CacheBlockReqest& cb_req,
+                              io::FileBlockSPtr& fb, const io::UInt128Wrapper& hash,
+                              size_t file_size, bool do_fill, butil::IOBuf* response_attachment,
+                              PFetchPeerDataResponse* response) {
+    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) {
+            return reject_peer_block_not_found(
+                    response, fmt::format("cache block not downloaded, {}",
+                                          format_peer_cache_block_context(request, cb_req, fb, hash,
+                                                                          file_size, do_fill)));
+        }
+        // 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),
+                get_peer_cache_filename(request->path()),
+                get_rowset_meta_resource_id_from_request(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 reject_peer_block_not_found(
+                response, 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_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, response_attachment);
+    if (!read_status.ok()) {
+        set_error_response(response, "read cache file error");
+        return read_status;
+    }
+    return Status::OK();
+}
+
 Status handle_peer_file_cache_block_request(const PFetchPeerDataRequest* request,
                                             PFetchPeerDataResponse* response,
                                             brpc::Controller* cntl) {
@@ -722,6 +814,7 @@ Status handle_peer_file_cache_block_request(const PFetchPeerDataRequest* request
     const bool use_attachment =
             cntl != nullptr && request->has_support_attachment() && request->support_attachment();
     response->set_data_in_attachment(use_attachment);
+    butil::IOBuf* response_attachment = use_attachment ? &cntl->response_attachment() : nullptr;
     if (use_attachment) {
         g_file_cache_get_by_peer_attachment_response_num << 1;
     } else {
@@ -737,8 +830,8 @@ Status handle_peer_file_cache_block_request(const PFetchPeerDataRequest* request
         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.
+        // Clip tail requests before the cache lookup 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;
@@ -750,102 +843,41 @@ Status handle_peer_file_cache_block_request(const PFetchPeerDataRequest* request
                     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();
+        if (do_fill) {
+            // Pull-through fill needs the EMPTY cell from get_or_set: the fill download lands in
+            // that cell, and this handler reads it back once the download finishes.
+            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) {
+                RETURN_IF_ERROR(serve_peer_cache_block(request, cb_req, fb, hash, file_size,
+                                                       do_fill, response_attachment, response));
+                ++response_blocks;
+            }
+            continue;
+        }
 
-        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;
-            }
+        // A plain peer read only serves blocks that are already downloaded. This lookup inserts no
+        // EMPTY cells, takes no downloader ownership, and accepts any cached layout, including
+        // short cells left by a pull-through fill from a client with a smaller cache block size.
+        // It also refreshes LRU state for the returned blocks.
+        io::FileBlocks blocks;
+        bool fully_covered = false;
+        RETURN_IF_ERROR(cache->get_downloaded_blocks_if_fully_covered(hash, offset, size, ctx,
+                                                                      &blocks, &fully_covered));
+        if (!fully_covered) {
+            return reject_peer_block_not_found(
+                    response,
+                    fmt::format(
+                            "cache block not fully downloaded, req_block=[offset={}, size={}], {}",
+                            cb_req.block_offset(), cb_req.block_size(),
+                            format_peer_request_context(request, hash, file_size)));
+        }
+        for (auto& fb : blocks) {
+            RETURN_IF_ERROR(serve_peer_cache_block(request, cb_req, fb, hash, file_size, do_fill,
+                                                   response_attachment, response));
             ++response_blocks;
         }
     }

be/src/io/cache/block_file_cache.cpp

@@ -55,6 +55,7 @@
 #include "io/cache/remote_scan_cache_write_limiter.h"
 #include "runtime/runtime_profile.h"
 #include "util/concurrency_stats.h"
+#include "util/defer_op.h"
 #include "util/mem_info.h"
 #include "util/stack_util.h"
 #include "util/stopwatch.hpp"
@@ -812,6 +813,14 @@ Status BlockFileCache::get_downloaded_blocks_if_fully_covered(const UInt128Wrapp
         return Status::OK();
     }
 
+    // Async touch defers LRU moves; queue them after the cache lock is released, as get_or_set()
+    // does. Declared before the lock so it runs after the lock guard is destroyed.
+    std::vector<FileBlockSPtr> need_update_lru_blocks;
+    Defer queue_lru_updates {[&]() {
+        for (auto& block : need_update_lru_blocks) {
+            add_need_update_lru_block(std::move(block));
+        }
+    }};
     FileBlock::Range range(offset, offset + size - 1);
     std::lock_guard cache_lock(_mutex);
     auto it = _files.find(hash);
@@ -882,9 +891,13 @@ Status BlockFileCache::get_downloaded_blocks_if_fully_covered(const UInt128Wrapp
     if (!*fully_covered) {
         return Status::OK();
     }
+    const bool async_touch = config::enable_file_cache_async_touch_on_get_or_set;
     for (const auto* cell : covered_cells) {
-        use_cell(*cell, blocks, need_to_move(cell->file_block->cache_type(), context.cache_type),
-                 cache_lock);
+        const bool move_iter = need_to_move(cell->file_block->cache_type(), context.cache_type);
+        use_cell(*cell, blocks, move_iter, cache_lock);
+        if (async_touch && move_iter) {
+            need_update_lru_blocks.push_back(cell->file_block);
+        }
     }
     return Status::OK();
 }

be/src/io/cache/cached_remote_file_reader.cpp

@@ -104,6 +104,7 @@ bvar::PassiveStatus<int> g_active_peer_races_bvar(
 bvar::Adder<uint64_t> g_peer_race_peer_win("peer_race_peer_win");
 bvar::Adder<uint64_t> g_peer_race_s3_win("peer_race_s3_win");
 bvar::Adder<uint64_t> g_peer_race_both_fail("peer_race_both_fail");
+bvar::Adder<uint64_t> g_peer_race_s3_skipped_after_queue("peer_race_s3_skipped_after_queue");
 bvar::Adder<uint64_t> g_peer_cross_compute_group_read("peer_cross_compute_group_read");
 bvar::Adder<uint64_t> g_peer_same_compute_group_read("peer_same_compute_group_read");
 bvar::Adder<uint64_t> g_peer_lazy_fetch_triggered("peer_lazy_fetch_triggered");
@@ -465,6 +466,20 @@ void launch_s3_race(std::shared_ptr<RaceState> race, size_t empty_start, size_t
     // peer wins; passing nullptr is safe because S3FileReader::read_at_impl ignores it.
     auto do_s3_read = [race, empty_start, span_size, remote_reader, owner]() {
         (void)owner;
+        TEST_SYNC_POINT("CachedRemoteFileReader::_execute_winner_race::s3_task_start");
+        {
+            // The peer may win while this task waits in PeerRaceS3ThreadPool. Skip the GET
+            // before allocating the span buffer or counting an S3 read.
+            std::unique_lock<bthread::Mutex> lk(race->mtx);
+            if (race->winner == 0) {
+                race->s3_done = true;
+                race->s3_status =
+                        Status::InternalError<false>("skipped: peer won before s3 task started");
+                race->cv.notify_all();
+                g_peer_race_s3_skipped_after_queue << 1;
+                return;
+            }
+        }
         auto s3_buf = std::make_unique<char[]>(span_size);
         size_t read_size = span_size;
         s3_read_counter << 1;

be/test/io/cache/cached_remote_file_reader_peer_test.cpp

@@ -28,6 +28,7 @@
 #include <condition_variable>
 #include <filesystem>
 #include <fstream>
+#include <functional>
 #include <mutex>
 #include <string>
 #include <thread>
@@ -53,6 +54,7 @@
 
 namespace doris {
 extern bvar::Adder<uint64_t> g_file_cache_get_by_peer_queue_timeout_num;
+extern bvar::Adder<uint64_t> g_file_cache_get_by_peer_get_or_set_calls;
 
 Status test_handle_peer_file_cache_block_request(const PFetchPeerDataRequest* request,
                                                  PFetchPeerDataResponse* response,
@@ -64,6 +66,9 @@ bool test_try_reject_if_queue_timed_out(std::chrono::steady_clock::time_point en
 namespace doris::io {
 
 extern bvar::Adder<uint64_t> peer_cache_reader_read_counter;
+extern bvar::Adder<uint64_t> s3_read_counter;
+extern bvar::Adder<uint64_t> g_peer_race_peer_win;
+extern bvar::Adder<uint64_t> g_peer_race_s3_skipped_after_queue;
 
 namespace {
 
@@ -250,6 +255,9 @@ public:
                             : "";
         }
 
+        if (before_response) {
+            before_response();
+        }
         if (_options.fail_status) {
             response->mutable_status()->set_status_code(TStatusCode::INTERNAL_ERROR);
             return;
@@ -345,6 +353,8 @@ public:
     std::atomic<int> request_block_count {0};
     std::atomic<bool> support_attachment {false};
     std::atomic<bool> request_fill {false};
+    // Runs on the RPC thread before the response is built; tests use it to order the peer reply.
+    std::function<void()> before_response;
     std::string get_last_fill_resource_id() const {
         std::lock_guard lock(_request_mu);
         return last_fill_resource_id;
@@ -994,6 +1004,166 @@ TEST_F(CachedRemoteFileReaderPeerTest, peer_file_cache_handler_allows_cached_blo
             create_peer_cache_block_request(file_path, 0, kPeerTestBlockSize, content.size());
     PFetchPeerDataResponse response;
     brpc::Controller cntl;
+    const uint64_t get_or_set_before = doris::g_file_cache_get_by_peer_get_or_set_calls.get_value();
+
+    auto st = doris::test_handle_peer_file_cache_block_request(&request, &response, &cntl);
+
+    ASSERT_TRUE(st.ok()) << st;
+    EXPECT_EQ(response.status().status_code(), TStatusCode::OK);
+    ASSERT_EQ(response.datas_size(), 1);
+    EXPECT_EQ(response.datas(0).block_offset(), 0);
+    EXPECT_EQ(response.datas(0).data(), content.substr(0, kPeerTestBlockSize));
+    // A plain peer read is served by the read-only lookup and never reaches get_or_set().
+    EXPECT_EQ(doris::g_file_cache_get_by_peer_get_or_set_calls.get_value() - get_or_set_before, 0);
+}
+
+TEST_F(CachedRemoteFileReaderPeerTest, peer_file_cache_handler_serves_fragmented_cached_layout) {
+    const std::string content = "abcdefghijklmnop";
+    const fs::path file_path =
+            create_peer_test_file("cached_remote_reader_peer_fragmented_layout.dat", content);
+    Defer cleanup_file {[&]() {
+        std::error_code ec;
+        fs::remove(file_path, ec);
+    }};
+
+    const fs::path cache_path = caches_dir / "cached_remote_reader_peer_fragmented_layout_cache";
+    Defer cleanup_cache {[&]() {
+        std::error_code ec;
+        fs::remove_all(cache_path, ec);
+    }};
+
+    clear_cached_remote_reader_factory();
+    BlockFileCache* cache = create_peer_test_cache(cache_path, kPeerTestBlockSize);
+    // A pull-through fill from a client with a smaller cache block size leaves short cells inside
+    // one of this cache's blocks. A block-sized plain peer read must still be served.
+    populate_cache_block_with_content(cache, file_path, 0, kPeerTestBlockSize / 2, content);
+    populate_cache_block_with_content(cache, file_path, kPeerTestBlockSize / 2,
+                                      kPeerTestBlockSize / 2, content);
+
+    auto request =
+            create_peer_cache_block_request(file_path, 0, kPeerTestBlockSize, content.size());
+    PFetchPeerDataResponse response;
+    brpc::Controller cntl;
+
+    auto st = doris::test_handle_peer_file_cache_block_request(&request, &response, &cntl);
+
+    ASSERT_TRUE(st.ok()) << st;
+    EXPECT_EQ(response.status().status_code(), TStatusCode::OK);
+    ASSERT_EQ(response.datas_size(), 2);
+    EXPECT_EQ(response.datas(0).block_offset(), 0);
+    EXPECT_EQ(response.datas(0).data(), content.substr(0, kPeerTestBlockSize / 2));
+    EXPECT_EQ(response.datas(1).block_offset(), kPeerTestBlockSize / 2);
+    EXPECT_EQ(response.datas(1).data(),
+              content.substr(kPeerTestBlockSize / 2, kPeerTestBlockSize / 2));
+}
+
+TEST_F(CachedRemoteFileReaderPeerTest, peer_file_cache_handler_queues_async_lru_touch) {
+    const bool old_async_touch = config::enable_file_cache_async_touch_on_get_or_set;
+    const int64_t old_update_interval_ms =
+            config::file_cache_background_block_lru_update_interval_ms;
+    Defer restore_config {[&]() {
+        config::enable_file_cache_async_touch_on_get_or_set = old_async_touch;
+        config::file_cache_background_block_lru_update_interval_ms = old_update_interval_ms;
+    }};
+    config::enable_file_cache_async_touch_on_get_or_set = true;
+    // Keep the background LRU worker from draining the queue before it is inspected.
+    config::file_cache_background_block_lru_update_interval_ms = 60 * 60 * 1000;
+
+    const std::string content = "abcdefghijklmnop";
+    const fs::path file_path =
+            create_peer_test_file("cached_remote_reader_peer_async_lru_touch.dat", content);
+    Defer cleanup_file {[&]() {
+        std::error_code ec;
+        fs::remove(file_path, ec);
+    }};
+
+    const fs::path cache_path = caches_dir / "cached_remote_reader_peer_async_lru_touch_cache";
+    Defer cleanup_cache {[&]() {
+        std::error_code ec;
+        fs::remove_all(cache_path, ec);
+    }};
+
+    clear_cached_remote_reader_factory();
+    BlockFileCache* cache = create_peer_test_cache(cache_path, kPeerTestBlockSize);
+    populate_cache_block_with_content(cache, file_path, 0, kPeerTestBlockSize, content);
+
+    auto request =
+            create_peer_cache_block_request(file_path, 0, kPeerTestBlockSize, content.size());
+    PFetchPeerDataResponse response;
+    brpc::Controller cntl;
+    const size_t pending_before = cache->need_update_lru_blocks_size_unsafe();
+
+    auto st = doris::test_handle_peer_file_cache_block_request(&request, &response, &cntl);
+
+    ASSERT_TRUE(st.ok()) << st;
+    ASSERT_EQ(response.datas_size(), 1);
+    // Async touch defers the LRU move, so a served hit must be queued like a get_or_set() hit.
+    EXPECT_EQ(cache->need_update_lru_blocks_size_unsafe(), pending_before + 1);
+}
+
+TEST_F(CachedRemoteFileReaderPeerTest, peer_file_cache_handler_miss_returns_not_found) {
+    const std::string content = "abcdefghijklmnop";
+    const fs::path file_path = create_peer_test_file("cached_remote_reader_peer_miss.dat", content);
+    Defer cleanup_file {[&]() {
+        std::error_code ec;
+        fs::remove(file_path, ec);
+    }};
+
+    const fs::path cache_path = caches_dir / "cached_remote_reader_peer_miss_cache";
+    Defer cleanup_cache {[&]() {
+        std::error_code ec;
+        fs::remove_all(cache_path, ec);
+    }};
+
+    clear_cached_remote_reader_factory();
+    BlockFileCache* cache = create_peer_test_cache(cache_path, kPeerTestBlockSize);
+    const auto hash = BlockFileCache::hash(file_path.filename().native());
+
+    auto request =
+            create_peer_cache_block_request(file_path, 0, kPeerTestBlockSize, content.size());
+    PFetchPeerDataResponse response;
+    brpc::Controller cntl;
+    const uint64_t get_or_set_before = doris::g_file_cache_get_by_peer_get_or_set_calls.get_value();
+
+    auto st = doris::test_handle_peer_file_cache_block_request(&request, &response, &cntl);
+
+    ASSERT_TRUE(st.is<ErrorCode::NOT_FOUND>()) << st;
+    EXPECT_EQ(response.status().status_code(), TStatusCode::NOT_FOUND);
+    ASSERT_GE(response.status().error_msgs_size(), 1);
+    EXPECT_NE(response.status().error_msgs(0).find("cache block not fully downloaded"),
+              std::string::npos);
+    EXPECT_EQ(response.datas_size(), 0);
+    EXPECT_EQ(doris::g_file_cache_get_by_peer_get_or_set_calls.get_value() - get_or_set_before, 0);
+    EXPECT_FALSE(cache->_files.contains(hash));
+}
+
+TEST_F(CachedRemoteFileReaderPeerTest, peer_file_cache_handler_serves_offset_inside_cached_block) {
+    const std::string content = "abcdefghijklmnop";
+    const fs::path file_path =
+            create_peer_test_file("cached_remote_reader_peer_offset_inside_block.dat", content);
+    Defer cleanup_file {[&]() {
+        std::error_code ec;
+        fs::remove(file_path, ec);
+    }};
+
+    const fs::path cache_path = caches_dir / "cached_remote_reader_peer_offset_inside_block_cache";
+    Defer cleanup_cache {[&]() {
+        std::error_code ec;
+        fs::remove_all(cache_path, ec);
+    }};
+
+    clear_cached_remote_reader_factory();
+    BlockFileCache* cache = create_peer_test_cache(cache_path, kPeerTestBlockSize);
+    populate_cache_block_with_content(cache, file_path, 0, kPeerTestBlockSize, content);
+
+    // A client with a smaller cache block size can request a range that starts inside one of this
+    // cache's blocks. The server returns the covering block whole. This checks the server only:
+    // the current client rejects a response block outside its requested ranges, as with
+    // get_or_set().
+    auto request = create_peer_cache_block_request(file_path, kPeerTestBlockSize / 2,
+                                                   kPeerTestBlockSize / 2, content.size());
+    PFetchPeerDataResponse response;
+    brpc::Controller cntl;
 
     auto st = doris::test_handle_peer_file_cache_block_request(&request, &response, &cntl);
 
@@ -2260,6 +2430,107 @@ protected:
                 .get_peer_candidates(kCrossTabletId);
     }
 
+    // Let the peer win while the S3 task is held at `s3_sync_point`, wait for the S3 task to
+    // finish, and report how many S3 reads and queue skips it recorded. With
+    // `peer_waits_for_s3_hold`, the peer replies only after S3 reaches the sync point, so the
+    // peer cannot win before the S3 task starts.
+    void run_peer_win_with_s3_held_at(const std::string& name, const std::string& s3_sync_point,
+                                      bool peer_waits_for_s3_hold, uint64_t* s3_reads,
+                                      uint64_t* s3_skips) {
+        const std::string content = "abcdefghijklmnop";
+        const fs::path file_path = create_cross_cg_test_file(name + "_0.dat", content);
+        Defer cleanup_file {[&]() {
+            std::error_code ec;
+            fs::remove(file_path, ec);
+        }};
+
+        const fs::path cache_path = caches_dir / (name + "_cache");
+        Defer cleanup_cache {[&]() {
+            std::error_code ec;
+            fs::remove_all(cache_path, ec);
+        }};
+
+        clear_cached_remote_reader_factory();
+        create_peer_test_cache(cache_path, kPeerTestBlockSize);
+
+        // Submit S3 immediately so only the S3 task itself can observe the peer win.
+        const int32_t old_hedge_delay_ms = config::peer_race_hedge_delay_ms;
+        Defer restore_hedge_delay {
+                [&]() { config::peer_race_hedge_delay_ms = old_hedge_delay_ms; }};
+        config::peer_race_hedge_delay_ms = 0;
+
+        std::mutex s3_hold_mutex;
+        std::condition_variable s3_hold_cv;
+        bool s3_held = false;
+        MockPeerCacheService service(content);
+        if (peer_waits_for_s3_hold) {
+            service.before_response = [&]() {
+                std::unique_lock lock(s3_hold_mutex);
+                s3_hold_cv.wait_for(lock, kPeerRpcWaitTimeout, [&] { return s3_held; });
+            };
+        }
+        brpc::Server server;
+        auto addr = start_peer_test_server(&server, &service);
+        Defer stop_server {[&]() { stop_peer_test_server(&server); }};
+        register_cross_cg_candidate("127.0.0.1", addr.port);
+
+        const uint64_t peer_wins_before = g_peer_race_peer_win.get_value();
+        const uint64_t s3_reads_before = s3_read_counter.get_value();
+        const uint64_t s3_skips_before = g_peer_race_s3_skipped_after_queue.get_value();
+
+        auto* sp = SyncPoint::get_instance();
+        sp->clear_all_call_backs();
+        sp->enable_processing();
+        SyncPoint::CallbackGuard s3_hold_guard;
+        sp->set_call_back(
+                s3_sync_point,
+                [&](auto&&) {
+                    {
+                        std::lock_guard lock(s3_hold_mutex);
+                        s3_held = true;
+                    }
+                    s3_hold_cv.notify_all();
+                    // collect_race_result counts the peer win only after winner == 0 is set.
+                    const auto deadline = std::chrono::steady_clock::now() + kPeerRpcWaitTimeout;
+                    while (g_peer_race_peer_win.get_value() == peer_wins_before &&
+                           std::chrono::steady_clock::now() < deadline) {
+                        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+                    }
+                },
+                &s3_hold_guard);
+        Defer disable_sp {[&]() {
+            sp->disable_processing();
+            sp->clear_all_call_backs();
+        }};
+
+        FileReaderSPtr local_reader;
+        ASSERT_TRUE(global_local_filesystem()->open_file(file_path.string(), &local_reader).ok());
+
+        FileReaderOptions opts;
+        opts.cache_type = FileCachePolicy::FILE_BLOCK_CACHE;
+        opts.is_doris_table = true;
+        opts.mtime = 1;
+        opts.tablet_id = kCrossTabletId;
+        auto reader = std::make_shared<CachedRemoteFileReader>(local_reader, opts);
+
+        std::string buffer(10, '#');
+        size_t bytes_read = 0;
+        IOContext io_ctx;
+        FileCacheStatistics cache_stats;
+        io_ctx.file_cache_stats = &cache_stats;
+
+        ASSERT_TRUE(
+                reader->read_at(1, Slice(buffer.data(), buffer.size()), &bytes_read, &io_ctx).ok());
+        EXPECT_EQ(buffer, content.substr(1, 10));
+        EXPECT_EQ(bytes_read, 10);
+        EXPECT_EQ(cache_stats.num_peer_race_peer_win, 1);
+
+        // The losing S3 task outlives read_at(); wait for it before reading its counters.
+        ExecEnv::GetInstance()->peer_race_s3_thread_pool()->wait();
+        *s3_reads = s3_read_counter.get_value() - s3_reads_before;
+        *s3_skips = g_peer_race_s3_skipped_after_queue.get_value() - s3_skips_before;
+    }
+
 private:
     bool _old_enable_peer_s3_race = false;
     int32_t _old_max_concurrent_peer_races = 64;
@@ -3418,6 +3689,38 @@ TEST_F(CrossCGWinnerRaceTest, race_peer_stops_when_s3_wins) {
     EXPECT_EQ(miss_service2.rpc_count.load(), 0) << "candidate 2 should be skipped (S3 won)";
 }
 
+// ----------------------------------------------------------------
+// Test: an S3 task that starts after the peer has won skips its GET.
+//
+// Holding S3 at s3_task_start models a task that waited in PeerRaceS3ThreadPool while the peer
+// won. The task must return before counting or issuing the S3 read.
+// ----------------------------------------------------------------
+TEST_F(CrossCGWinnerRaceTest, s3_task_skips_read_when_peer_won_while_queued) {
+    uint64_t s3_reads = 0;
+    uint64_t s3_skips = 0;
+    run_peer_win_with_s3_held_at("s3_skip_queued",
+                                 "CachedRemoteFileReader::_execute_winner_race::s3_task_start",
+                                 /*peer_waits_for_s3_hold=*/false, &s3_reads, &s3_skips);
+    EXPECT_EQ(s3_reads, 0);
+    EXPECT_EQ(s3_skips, 1);
+}
+
+// ----------------------------------------------------------------
+// Test: an S3 task that started before the peer won still reads.
+//
+// s3_before_read is after the queue check. The peer replies only once S3 is held there, so the
+// task has passed the check before the peer wins and keeps the read-then-discard behavior.
+// ----------------------------------------------------------------
+TEST_F(CrossCGWinnerRaceTest, s3_task_reads_when_started_before_peer_won) {
+    uint64_t s3_reads = 0;
+    uint64_t s3_skips = 0;
+    run_peer_win_with_s3_held_at("s3_read_started",
+                                 "CachedRemoteFileReader::_execute_winner_race::s3_before_read",
+                                 /*peer_waits_for_s3_hold=*/true, &s3_reads, &s3_skips);
+    EXPECT_EQ(s3_reads, 1);
+    EXPECT_EQ(s3_skips, 0);
+}
+
 // ----------------------------------------------------------------
 // Test: cross-CG peer wins race and timer metrics are non-zero.
 //