[BugFix]Fix filesystem cache failure. (#65823)

Signed-off-by: edwinhzhang <[email protected]>
Signed-off-by: edwinhzhang <[email protected]>
(cherry picked from commit 06a0a19)

Author: zhangheihei

be/src/service/staros_worker.cpp

@@ -202,9 +202,15 @@ absl::StatusOr<std::shared_ptr<fslib::FileSystem>> StarOSWorker::get_shard_files
             return build_filesystem_on_demand(id, conf);
         }
 
-        auto fs = lookup_fs_cache(it->second.fs_cache_key);
-        if (fs != nullptr) {
-            return fs;
+        // Cache miss: reset the fs_cache_key to ensure a new shared_ptr will be created
+        {
+            std::lock_guard<std::mutex> reset_lock(_fs_cache_key_reset_mtx);
+            auto fs = lookup_fs_cache(it->second.fs_cache_key);
+            if (fs != nullptr) {
+                return fs;
+            }
+
+            it->second.fs_cache_key.reset();
         }
         shard_info = it->second.shard_info;
     }

be/src/service/staros_worker.h

@@ -124,6 +124,7 @@ class StarOSWorker : public staros::starlet::Worker {
 private:
     mutable std::shared_mutex _mtx;
     std::shared_mutex _cache_mtx;
+    std::mutex _fs_cache_key_reset_mtx; // Protects fs_cache_key reset operations
     std::unordered_map<ShardId, ShardInfoDetails> _shards;
     std::unique_ptr<Cache> _fs_cache;
 };