[fix](cloud) refresh tablet meta of continuously ingested tablets

be/src/cloud/cloud_base_compaction.cpp

@@ -67,7 +67,7 @@ Status CloudBaseCompaction::prepare_compact() {
         // synchronized with meta-service.
         if (_tablet->tablet_meta()->all_rs_metas().size() >=
                     cloud_tablet()->fetch_add_approximate_num_rowsets(0) &&
-            cloud_tablet()->last_sync_time_s > 0) {
+            cloud_tablet()->last_sync_rowsets_time_s > 0) {
             need_sync_tablet = false;
         }
     }
@@ -136,7 +136,7 @@ Status CloudBaseCompaction::request_global_lock() {
         cloud_tablet()->set_last_base_compaction_failure_time(UnixMillis());
         if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
             // set last_sync_time to 0 to force sync tablet next time
-            cloud_tablet()->last_sync_time_s = 0;
+            cloud_tablet()->last_sync_rowsets_time_s = 0;
         } else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
             // tablet not found
             cloud_tablet()->clear_cache();
@@ -440,7 +440,7 @@ Status CloudBaseCompaction::modify_rowsets() {
         // the tablet to be unable to synchronize the rowset meta changes generated by cumu compaction.
         cloud_tablet()->set_base_compaction_cnt(stats.base_compaction_cnt());
         if (stats.cumulative_point() > cloud_tablet()->cumulative_layer_point()) {
-            cloud_tablet()->last_sync_time_s = 0;
+            cloud_tablet()->last_sync_rowsets_time_s = 0;
         }
         if (output_rowset_delete_bitmap) {
             _tablet->tablet_meta()->delete_bitmap().merge(*output_rowset_delete_bitmap);

be/src/cloud/cloud_cumulative_compaction.cpp

@@ -157,7 +157,7 @@ Status CloudCumulativeCompaction::prepare_compact() {
         // synchronized with meta-service.
         if (_tablet->tablet_meta()->all_rs_metas().size() >=
                     cloud_tablet()->fetch_add_approximate_num_rowsets(0) &&
-            cloud_tablet()->last_sync_time_s > 0) {
+            cloud_tablet()->last_sync_rowsets_time_s > 0) {
             need_sync_tablet = false;
         }
     }
@@ -238,7 +238,7 @@ Status CloudCumulativeCompaction::request_global_lock() {
     if (!st.ok()) {
         if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
             // set last_sync_time to 0 to force sync tablet next time
-            cloud_tablet()->last_sync_time_s = 0;
+            cloud_tablet()->last_sync_rowsets_time_s = 0;
         } else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
             // tablet not found
             cloud_tablet()->clear_cache();
@@ -925,7 +925,7 @@ void CloudCumulativeCompaction::update_cumulative_point(int64_t input_cumulative
     if (!st.ok()) {
         if (start_resp.status().code() == cloud::STALE_TABLET_CACHE) {
             // set last_sync_time to 0 to force sync tablet next time
-            cloud_tablet()->last_sync_time_s = 0;
+            cloud_tablet()->last_sync_rowsets_time_s = 0;
         } else if (start_resp.status().code() == cloud::TABLET_NOT_FOUND) {
             // tablet not found
             cloud_tablet()->clear_cache();

be/src/cloud/cloud_full_compaction.cpp

@@ -113,7 +113,7 @@ Status CloudFullCompaction::request_global_lock() {
     if (!st.ok()) {
         if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
             // set last_sync_time to 0 to force sync tablet next time
-            cloud_tablet()->last_sync_time_s = 0;
+            cloud_tablet()->last_sync_rowsets_time_s = 0;
         } else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
             // tablet not found
             cloud_tablet()->clear_cache();

be/src/cloud/cloud_index_change_compaction.cpp

@@ -170,7 +170,7 @@ Status CloudIndexChangeCompaction::request_global_lock(bool& should_skip_err) {
     if (!st.ok()) {
         if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
             // set last_sync_time to 0 to force sync tablet next time
-            cloud_tablet()->last_sync_time_s = 0;
+            cloud_tablet()->last_sync_rowsets_time_s = 0;
             should_skip_err = true;
         } else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
             // tablet not found
@@ -380,7 +380,7 @@ void CloudIndexChangeCompaction::_update_tablet_for_base_compaction(
         // the tablet to be unable to synchronize the rowset meta changes generated by cumu compaction.
         cloud_tablet()->set_base_compaction_cnt(stats.base_compaction_cnt());
         if (stats.cumulative_point() > cloud_tablet()->cumulative_layer_point()) {
-            cloud_tablet()->last_sync_time_s = 0;
+            cloud_tablet()->last_sync_rowsets_time_s = 0;
         }
         if (output_rowset_delete_bitmap) {
             _tablet->tablet_meta()->delete_bitmap().merge(*output_rowset_delete_bitmap);

be/src/cloud/cloud_meta_mgr.cpp

@@ -885,7 +885,7 @@ Status CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet,
         }
 
         int64_t now = duration_cast<seconds>(system_clock::now().time_since_epoch()).count();
-        tablet->last_sync_time_s = now;
+        tablet->last_sync_rowsets_time_s = now;
 
         if (sync_stats) {
             sync_stats->get_remote_rowsets_rpc_ns +=

be/src/cloud/cloud_tablet.cpp

@@ -1623,6 +1623,7 @@ Status CloudTablet::sync_meta() {
         }
     }
 
+    last_sync_tablet_meta_time_s = ::time(nullptr);
     return Status::OK();
 }
 

be/src/cloud/cloud_tablet.h

@@ -348,7 +348,14 @@ class CloudTablet final : public BaseTablet {
 
     const auto& rowset_map() const { return _rs_version_map; }
 
-    int64_t last_sync_time_s = 0;
+    // How long since this tablet's ROWSETS were pulled from MS. Only sync_rowsets() advances
+    // it, and only when it actually issues the RPC.
+    int64_t last_sync_rowsets_time_s = 0;
+    // How long since this tablet's META was pulled from MS, which is what carries properties
+    // such as the file cache TTL. Only sync_meta() advances it. Tracked separately on
+    // purpose: a tablet under continuous ingest keeps the rowsets clock permanently fresh,
+    // so gating meta work on that one starves the meta refresh entirely.
+    int64_t last_sync_tablet_meta_time_s = 0;
     int64_t last_load_time_ms = 0;
     int64_t last_base_compaction_success_time_ms = 0;
     int64_t last_cumu_compaction_success_time_ms = 0;

be/src/cloud/cloud_tablet_mgr.cpp

@@ -19,7 +19,10 @@
 
 #include <bthread/countdown_event.h>
 
+#include <algorithm>
 #include <chrono>
+#include <set>
+#include <utility>
 
 #include "cloud/cloud_cluster_info.h"
 #include "cloud/cloud_meta_mgr.h"
@@ -40,6 +43,8 @@ bvar::Adder<uint64_t> g_base_compaction_not_frozen_tablet_num(
         "base_compaction_not_frozen_tablet_num");
 bvar::Adder<uint64_t> g_cumu_compaction_not_frozen_tablet_num(
         "cumu_compaction_not_frozen_tablet_num");
+bvar::Adder<uint64_t> g_sync_tablets_meta_num("sync_tablets_meta_num");
+bvar::Adder<uint64_t> g_sync_tablets_rowsets_num("sync_tablets_rowsets_num");
 namespace {
 
 // port from
@@ -386,53 +391,93 @@ std::vector<std::weak_ptr<CloudTablet>> CloudTabletMgr::get_weak_tablets() {
 
 void CloudTabletMgr::sync_tablets(const CountDownLatch& stop_latch) {
     LOG_INFO("begin to sync tablets");
-    int64_t last_sync_time_bound = ::time(nullptr) - config::tablet_sync_interval_s;
 
-    auto weak_tablets = get_weak_tablets();
+    // A tablet carries two staleness clocks and each one gates a different RPC:
+    //
+    //   last_sync_rowsets_time_s
+    //       how long since we pulled this tablet's ROWSETS from MS. Only sync_rowsets()
+    //       advances it, and only when it actually issues the RPC -- a query whose requested
+    //       version we already hold returns early and leaves the clock untouched.
+    //
+    //   last_sync_tablet_meta_time_s
+    //       how long since we pulled this tablet's META from MS, which is what carries
+    //       properties such as the file cache TTL. Only sync_meta() advances it.
+    //
+    // They have to be read separately. A tablet under continuous ingest keeps the rowsets
+    // clock permanently fresh, so selecting meta work by it -- as this used to -- means such a
+    // tablet never has its meta refreshed at all, and it keeps serving whatever TTL it was
+    // built with.
+    const int64_t stale_before = ::time(nullptr) - config::tablet_sync_interval_s;
+
+    struct Work {
+        std::weak_ptr<CloudTablet> tablet;
+        bool needs_meta = false;
+        bool needs_rowsets = false;
+    };
 
-    // sort by last_sync_time
+    // Ordered by the older of the two clocks, so that if we are told to stop half way, the
+    // tablets that have been waiting longest have already been served.
     static auto cmp = [](const auto& a, const auto& b) { return a.first < b.first; };
-    std::multiset<std::pair<int64_t, std::weak_ptr<CloudTablet>>, decltype(cmp)>
-            sync_time_tablet_set(cmp);
+    std::multiset<std::pair<int64_t, Work>, decltype(cmp)> due(cmp);
 
-    for (auto& weak_tablet : weak_tablets) {
-        if (auto tablet = weak_tablet.lock()) {
-            int64_t last_sync_time = tablet->last_sync_time_s;
-            if (last_sync_time <= last_sync_time_bound) {
-                sync_time_tablet_set.emplace(last_sync_time, weak_tablet);
-            }
+    for (auto& weak_tablet : get_weak_tablets()) {
+        auto tablet = weak_tablet.lock();
+        if (!tablet) {
+            continue;
+        }
+        const bool needs_rowsets = tablet->last_sync_rowsets_time_s <= stale_before;
+        Work work {
+                .tablet = weak_tablet,
+                // Pulling rowsets implies pulling the tablet meta: the rowsets we are about
+                // to take are only as trustworthy as the meta they belong to, and this is
+                // the relationship the previous single pass had.
+                .needs_meta = needs_rowsets || tablet->last_sync_tablet_meta_time_s <= stale_before,
+                .needs_rowsets = needs_rowsets};
+        if (!work.needs_meta && !work.needs_rowsets) {
+            continue;
         }
+        due.emplace(
+                std::min(tablet->last_sync_tablet_meta_time_s, tablet->last_sync_rowsets_time_s),
+                std::move(work));
     }
 
     int num_sync = 0;
-    for (auto&& [_, weak_tablet] : sync_time_tablet_set) {
+    int num_sync_meta = 0;
+    for (auto&& [_, work] : due) {
         if (stop_latch.count() <= 0) {
             break;
         }
+        auto tablet = work.tablet.lock();
+        if (!tablet) {
+            continue;
+        }
 
-        if (auto tablet = weak_tablet.lock()) {
-            if (tablet->last_sync_time_s > last_sync_time_bound) {
-                continue;
-            }
-
-            ++num_sync;
+        if (work.needs_meta) {
+            ++num_sync_meta;
+            g_sync_tablets_meta_num << 1;
             auto st = tablet->sync_meta();
             if (!st) {
                 LOG_WARNING("failed to sync tablet meta {}", tablet->tablet_id()).error(st);
                 if (st.is<ErrorCode::NOT_FOUND>()) {
+                    // the tablet is gone from MS, there is nothing left to sync
                     continue;
                 }
             }
+        }
+
+        if (work.needs_rowsets) {
+            ++num_sync;
+            g_sync_tablets_rowsets_num << 1;
             SyncOptions options;
             options.query_version = -1;
             options.merge_schema = true;
-            st = tablet->sync_rowsets(options);
+            auto st = tablet->sync_rowsets(options);
             if (!st) {
                 LOG_WARNING("failed to sync tablet rowsets {}", tablet->tablet_id()).error(st);
             }
         }
     }
-    LOG_INFO("finish sync tablets").tag("num_sync", num_sync);
+    LOG_INFO("finish sync tablets").tag("num_sync", num_sync).tag("num_sync_meta", num_sync_meta);
 }
 
 Status CloudTabletMgr::get_topn_tablets_to_compact(

be/src/storage/compaction/compaction.cpp

@@ -2112,7 +2112,7 @@ bool CloudCompactionMixin::should_apply_cumulative_compaction_result(
     }
     if (response_cumulative_compaction_cnt != local_cumulative_compaction_cnt + 1) {
         // Only the current task's output is available locally. Sync all missing outputs instead.
-        cloud_tablet()->last_sync_time_s = 0;
+        cloud_tablet()->last_sync_rowsets_time_s = 0;
         LOG_INFO("defer applying cumulative compaction result until tablet sync")
                 .tag("tablet_id", _tablet->tablet_id())
                 .tag("job_id", _uuid)

be/test/cloud/cloud_compaction_test.cpp

@@ -456,7 +456,7 @@ TEST_F(CloudCompactionTest, cumulative_global_lock_failure_keeps_thread_count_ba
         std::unique_lock lock(tablet->get_header_lock());
         tablet->add_rowsets(std::move(rowsets), false, lock);
     }
-    tablet->last_sync_time_s = 1;
+    tablet->last_sync_rowsets_time_s = 1;
     tablet->_approximate_num_rowsets = 0;
 
     ASSERT_TRUE(ThreadPoolBuilder("CumuCompactionTaskThreadPoolTest")
@@ -579,16 +579,16 @@ class TestableCloudCompaction : public CloudCompactionMixin {
 TEST_F(CloudCompactionTest, cumulative_result_requires_next_counter) {
     auto tablet = std::make_shared<CloudTablet>(_engine, _tablet_meta);
     tablet->set_cumulative_compaction_cnt(1);
-    tablet->last_sync_time_s = 1;
+    tablet->last_sync_rowsets_time_s = 1;
     TestableCloudCompaction compaction(_engine, tablet);
 
     std::unique_lock lock(tablet->get_header_lock());
     EXPECT_FALSE(compaction.test_should_apply_cumulative_compaction_result(1));
-    EXPECT_EQ(tablet->last_sync_time_s, 1);
+    EXPECT_EQ(tablet->last_sync_rowsets_time_s, 1);
     EXPECT_TRUE(compaction.test_should_apply_cumulative_compaction_result(2));
-    EXPECT_EQ(tablet->last_sync_time_s, 1);
+    EXPECT_EQ(tablet->last_sync_rowsets_time_s, 1);
     EXPECT_FALSE(compaction.test_should_apply_cumulative_compaction_result(3));
-    EXPECT_EQ(tablet->last_sync_time_s, 0);
+    EXPECT_EQ(tablet->last_sync_rowsets_time_s, 0);
 }
 
 class TestableCloudCumulativeCompaction : public CloudCumulativeCompaction {
@@ -627,7 +627,7 @@ static CloudTabletSPtr create_cloud_tablet_with_rowsets(CloudStorageEngine& engi
     tablet->set_cumulative_layer_point(cumulative_point);
     tablet->fetch_add_approximate_num_rowsets(static_cast<int64_t>(num_rowsets) -
                                               tablet->fetch_add_approximate_num_rowsets(0));
-    tablet->last_sync_time_s = 1;
+    tablet->last_sync_rowsets_time_s = 1;
     return tablet;
 }
 
@@ -694,7 +694,7 @@ TEST_F(CloudCompactionTest, base_result_with_newer_cumulative_point_forces_sync)
 
         ASSERT_TRUE(compaction.modify_rowsets().ok());
         EXPECT_EQ(tablet->cumulative_layer_point(), 6);
-        EXPECT_EQ(tablet->last_sync_time_s, expected_sync_time);
+        EXPECT_EQ(tablet->last_sync_rowsets_time_s, expected_sync_time);
     };
 
     run_case(10008, 8, 0);
@@ -903,7 +903,7 @@ TEST_F(CloudCompactionTest, parallel_pick_keeps_mode_after_dynamic_config_change
     EXPECT_TRUE(commit_called);
     EXPECT_EQ(tablet->cumulative_compaction_cnt(), 0);
     EXPECT_EQ(tablet->cumulative_layer_point(), 2);
-    EXPECT_EQ(tablet->last_sync_time_s, 0);
+    EXPECT_EQ(tablet->last_sync_rowsets_time_s, 0);
 }
 
 TEST_F(CloudCompactionTest, parallel_pick_advances_continuous_low_prefix_through_delete) {