[fix](filecache) self-heal stale DOWNLOADED entries on local NOT_FOUND (#60977)

freemandealer · web-flow · commit ca93e7baa551 · 2026-03-06T15:40:15.000+08:00
Problem:
In a rare restart window, BE can rebuild file-cache metadata in memory
while
the corresponding cache files are not yet durable on disk. If that
metadata is
also restored via LRU dump/load, blocks may appear as DOWNLOADED even
though
the local files are missing. Subsequent reads then produce
false-positive cache
hits, fail on local read, and repeatedly fall back to S3. This preserves
  correctness but causes avoidable cache thrashing and latency jitter.

Root cause:
The read path treated DOWNLOADED as a valid local hit source and fell
back to
remote reads on failure, but it did not actively invalidate stale
metadata when
  the local cache file was gone.
diff --git a/be/src/io/cache/cached_remote_file_reader.cpp b/be/src/io/cache/cached_remote_file_reader.cpp
@@ -81,6 +81,8 @@ bvar::Adder<uint64_t> g_read_cache_direct_partial_bytes(
 bvar::Adder<uint64_t> g_read_cache_indirect_bytes("cached_remote_reader_cache_indirect_bytes");
 bvar::Adder<uint64_t> g_read_cache_indirect_total_bytes(
         "cached_remote_reader_cache_indirect_total_bytes");
+bvar::Adder<uint64_t> g_read_cache_self_heal_on_not_found(
+        "cached_remote_reader_self_heal_on_not_found");
 bvar::Window<bvar::Adder<uint64_t>> g_read_cache_indirect_bytes_1min_window(
         "cached_remote_reader_indirect_bytes_1min_window", &g_read_cache_indirect_bytes, 60);
 bvar::Window<bvar::Adder<uint64_t>> g_read_cache_indirect_total_bytes_1min_window(
@@ -488,6 +490,7 @@ Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t*
 
     size_t current_offset = offset;
     size_t end_offset = offset + bytes_req - 1;
+    bool need_self_heal = false;
     *bytes_read = 0;
     for (auto& block : holder.file_blocks) {
         if (current_offset > end_offset) {
@@ -549,6 +552,17 @@ Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t*
                 if (is_dryrun) [[unlikely]] {
                     // dryrun mode uses a null buffer, skip actual remote IO
                 } else {
+                    if (block_state == FileBlock::State::DOWNLOADED &&
+                        st.is<ErrorCode::NOT_FOUND>()) {
+                        need_self_heal = true;
+                        g_read_cache_self_heal_on_not_found << 1;
+                        LOG_EVERY_N(WARNING, 100)
+                                << "Cache block file is missing, will self-heal by clearing cache "
+                                   "hash. "
+                                << "path=" << path().native()
+                                << ", hash=" << _cache_hash.to_string() << ", offset=" << left
+                                << ", err=" << st.msg();
+                    }
                     LOG(WARNING) << "Read data failed from file cache downloaded by others. err="
                                  << st.msg() << ", block state=" << block_state;
                     size_t nest_bytes_read {0};
@@ -568,6 +582,9 @@ Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result, size_t*
         *bytes_read += read_size;
         current_offset = right + 1;
     }
+    if (need_self_heal && _cache != nullptr) {
+        _cache->remove_if_cached_async(_cache_hash);
+    }
     g_read_cache_indirect_bytes << indirect_read_bytes;
     g_read_cache_indirect_total_bytes << *bytes_read;
 
diff --git a/be/test/io/cache/block_file_cache_test.cpp b/be/test/io/cache/block_file_cache_test.cpp
@@ -3566,6 +3566,191 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_error_handle) {
     FileCacheFactory::instance()->_capacity = 0;
 }
 
+extern bvar::Adder<uint64_t> g_read_cache_self_heal_on_not_found;
+
+TEST_F(BlockFileCacheTest, cached_remote_file_reader_self_heal_on_downloaded_not_found) {
+    bool origin_enable_direct_read = config::enable_read_cache_file_directly;
+    config::enable_read_cache_file_directly = false;
+    Defer reset_direct_read {
+            [&] { config::enable_read_cache_file_directly = origin_enable_direct_read; }};
+
+    std::string cache_base_path =
+            caches_dir / "cached_remote_reader_self_heal_on_downloaded_not_found" / "";
+    if (fs::exists(cache_base_path)) {
+        fs::remove_all(cache_base_path);
+    }
+    fs::create_directories(cache_base_path);
+
+    io::FileCacheSettings settings;
+    settings.query_queue_size = 6291456;
+    settings.query_queue_elements = 6;
+    settings.index_queue_size = 1048576;
+    settings.index_queue_elements = 1;
+    settings.disposable_queue_size = 1048576;
+    settings.disposable_queue_elements = 1;
+    settings.capacity = 8388608;
+    settings.max_file_block_size = 1048576;
+    settings.max_query_cache_size = 0;
+    ASSERT_TRUE(FileCacheFactory::instance()->create_file_cache(cache_base_path, settings).ok());
+    auto cache = FileCacheFactory::instance()->_path_to_cache[cache_base_path];
+    for (int i = 0; i < 100; i++) {
+        if (cache->get_async_open_success()) {
+            break;
+        }
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+
+    FileReaderSPtr local_reader;
+    ASSERT_TRUE(global_local_filesystem()->open_file(tmp_file, &local_reader));
+    io::FileReaderOptions opts;
+    opts.cache_type = io::cache_type_from_string("file_block_cache");
+    opts.is_doris_table = true;
+    CachedRemoteFileReader reader(local_reader, opts);
+
+    uint64_t before_self_heal = g_read_cache_self_heal_on_not_found.get_value();
+
+    std::string buffer(64_kb, '\0');
+    IOContext io_ctx;
+    FileCacheStatistics stats;
+    io_ctx.file_cache_stats = &stats;
+    size_t bytes_read = 0;
+    ASSERT_TRUE(reader.read_at(0, Slice(buffer.data(), buffer.size()), &bytes_read, &io_ctx).ok());
+    EXPECT_EQ(std::string(64_kb, '0'), buffer);
+
+    auto key = io::BlockFileCache::hash("tmp_file");
+    {
+        io::CacheContext inspect_ctx;
+        ReadStatistics inspect_stats;
+        inspect_ctx.stats = &inspect_stats;
+        inspect_ctx.cache_type = io::FileCacheType::NORMAL;
+        auto inspect_holder = cache->get_or_set(key, 0, 64_kb, inspect_ctx);
+        auto inspect_blocks = fromHolder(inspect_holder);
+        ASSERT_EQ(inspect_blocks.size(), 1);
+        ASSERT_EQ(inspect_blocks[0]->state(), io::FileBlock::State::DOWNLOADED);
+        std::string cache_file = inspect_blocks[0]->get_cache_file();
+        ASSERT_TRUE(fs::exists(cache_file));
+        ASSERT_TRUE(global_local_filesystem()->delete_file(cache_file).ok());
+        ASSERT_FALSE(fs::exists(cache_file));
+    }
+
+    ASSERT_TRUE(reader.read_at(0, Slice(buffer.data(), buffer.size()), &bytes_read, &io_ctx).ok());
+    EXPECT_EQ(std::string(64_kb, '0'), buffer);
+
+    bool self_healed = false;
+    for (int i = 0; i < 100; ++i) {
+        io::CacheContext verify_ctx;
+        ReadStatistics verify_stats;
+        verify_ctx.stats = &verify_stats;
+        verify_ctx.cache_type = io::FileCacheType::NORMAL;
+        auto verify_holder = cache->get_or_set(key, 0, 64_kb, verify_ctx);
+        auto verify_blocks = fromHolder(verify_holder);
+        if (verify_blocks.size() == 1 && verify_blocks[0]->state() == io::FileBlock::State::EMPTY) {
+            self_healed = true;
+            break;
+        }
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+    EXPECT_TRUE(self_healed);
+    EXPECT_EQ(g_read_cache_self_heal_on_not_found.get_value(), before_self_heal + 1);
+
+    EXPECT_TRUE(reader.close().ok());
+    EXPECT_TRUE(reader.closed());
+    std::this_thread::sleep_for(std::chrono::seconds(1));
+    if (fs::exists(cache_base_path)) {
+        fs::remove_all(cache_base_path);
+    }
+    FileCacheFactory::instance()->_caches.clear();
+    FileCacheFactory::instance()->_path_to_cache.clear();
+    FileCacheFactory::instance()->_capacity = 0;
+}
+
+TEST_F(BlockFileCacheTest, cached_remote_file_reader_no_self_heal_on_non_not_found_error) {
+    bool origin_enable_direct_read = config::enable_read_cache_file_directly;
+    config::enable_read_cache_file_directly = false;
+    Defer reset_direct_read {
+            [&] { config::enable_read_cache_file_directly = origin_enable_direct_read; }};
+
+    std::string cache_base_path =
+            caches_dir / "cached_remote_reader_no_self_heal_on_non_not_found_error" / "";
+    if (fs::exists(cache_base_path)) {
+        fs::remove_all(cache_base_path);
+    }
+    fs::create_directories(cache_base_path);
+
+    io::FileCacheSettings settings;
+    settings.query_queue_size = 6291456;
+    settings.query_queue_elements = 6;
+    settings.index_queue_size = 1048576;
+    settings.index_queue_elements = 1;
+    settings.disposable_queue_size = 1048576;
+    settings.disposable_queue_elements = 1;
+    settings.capacity = 8388608;
+    settings.max_file_block_size = 1048576;
+    settings.max_query_cache_size = 0;
+    ASSERT_TRUE(FileCacheFactory::instance()->create_file_cache(cache_base_path, settings).ok());
+    auto cache = FileCacheFactory::instance()->_path_to_cache[cache_base_path];
+    for (int i = 0; i < 100; i++) {
+        if (cache->get_async_open_success()) {
+            break;
+        }
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+
+    FileReaderSPtr local_reader;
+    ASSERT_TRUE(global_local_filesystem()->open_file(tmp_file, &local_reader));
+    io::FileReaderOptions opts;
+    opts.cache_type = io::cache_type_from_string("file_block_cache");
+    opts.is_doris_table = true;
+    CachedRemoteFileReader reader(local_reader, opts);
+
+    std::string buffer(64_kb, '\0');
+    IOContext io_ctx;
+    FileCacheStatistics stats;
+    io_ctx.file_cache_stats = &stats;
+    size_t bytes_read = 0;
+    ASSERT_TRUE(reader.read_at(0, Slice(buffer.data(), buffer.size()), &bytes_read, &io_ctx).ok());
+
+    uint64_t before_self_heal = g_read_cache_self_heal_on_not_found.get_value();
+    auto* sp = SyncPoint::get_instance();
+    sp->enable_processing();
+    Defer defer {[&] {
+        sp->clear_call_back("LocalFileReader::read_at_impl");
+        sp->disable_processing();
+    }};
+    sp->set_call_back("LocalFileReader::read_at_impl", [&](auto&& values) {
+        std::pair<Status, bool>* pair = try_any_cast<std::pair<Status, bool>*>(values.back());
+        pair->first = Status::IOError("inject io error for cache read");
+        pair->second = true;
+    });
+
+    auto st = reader.read_at(0, Slice(buffer.data(), buffer.size()), &bytes_read, &io_ctx);
+    ASSERT_FALSE(st.ok());
+    EXPECT_EQ(g_read_cache_self_heal_on_not_found.get_value(), before_self_heal);
+
+    sp->clear_call_back("LocalFileReader::read_at_impl");
+    sp->disable_processing();
+
+    io::CacheContext verify_ctx;
+    ReadStatistics verify_stats;
+    verify_ctx.stats = &verify_stats;
+    verify_ctx.cache_type = io::FileCacheType::NORMAL;
+    auto key = io::BlockFileCache::hash("tmp_file");
+    auto verify_holder = cache->get_or_set(key, 0, 64_kb, verify_ctx);
+    auto verify_blocks = fromHolder(verify_holder);
+    ASSERT_EQ(verify_blocks.size(), 1);
+    EXPECT_EQ(verify_blocks[0]->state(), io::FileBlock::State::DOWNLOADED);
+
+    EXPECT_TRUE(reader.close().ok());
+    EXPECT_TRUE(reader.closed());
+    std::this_thread::sleep_for(std::chrono::seconds(1));
+    if (fs::exists(cache_base_path)) {
+        fs::remove_all(cache_base_path);
+    }
+    FileCacheFactory::instance()->_caches.clear();
+    FileCacheFactory::instance()->_path_to_cache.clear();
+    FileCacheFactory::instance()->_capacity = 0;
+}
+
 TEST_F(BlockFileCacheTest, cached_remote_file_reader_init) {
     std::string cache_base_path = caches_dir / "cached_remote_file_reader_init" / "";
     if (fs::exists(cache_base_path)) {
