ct/housekeeper: don't RPC if start offset hasn't changed

Noticed that set_start_offset RPCs were constantly being sent to the
metastore, even for topics with a start offset had been 0 for a while.
Functionally this isn't problematic, but it's noise to the cluster.

This commit caches the last synced start offset so we can avoid the RPC.
It's transient, so when the housekeeper is reset (e.g. on partition
leadership change) we'll still see a potentially "wasted" no-op L1
set_start_offset call, but that's still better than what we have now.

Author: andrwng

src/v/cloud_topics/housekeeper/housekeeper.cc

@@ -190,6 +190,9 @@ ss::future<> housekeeper::sync_start_offset() {
         co_return;
     }
     auto start_offset = _l0_metastore->get_start_offset(_tidp);
+    if (_last_synced_start_offset == start_offset) {
+        co_return;
+    }
     vlog(cd_log.debug, "Setting {} start offset to {}", _tidp, start_offset);
     auto result = co_await _l1_metastore->set_start_offset(_tidp, start_offset);
     if (!result.has_value()) {
@@ -198,7 +201,9 @@ ss::future<> housekeeper::sync_start_offset() {
           "Failed to sync start offset to L1 for {}: {}",
           _tidp,
           result.error());
+        co_return;
     }
+    _last_synced_start_offset = start_offset;
 }
 
 } // namespace cloud_topics

src/v/cloud_topics/housekeeper/housekeeper.h

@@ -144,6 +144,7 @@ class housekeeper {
     ss::abort_source _as;
 
     std::optional<cloud_topics::cluster_epoch> _last_epoch;
+    std::optional<kafka::offset> _last_synced_start_offset;
 };
 
 } // namespace cloud_topics

src/v/cloud_topics/housekeeper/tests/housekeeper_test.cc

@@ -197,6 +197,17 @@ T handle_error(
         std::to_underlying(result.error())));
 }
 
+class counting_metastore : public cloud_topics::l1::simple_metastore {
+public:
+    ss::future<std::expected<void, errc>> set_start_offset(
+      const model::topic_id_partition& tidp, kafka::offset o) override {
+        ++set_start_offset_calls;
+        return simple_metastore::set_start_offset(tidp, o);
+    }
+
+    size_t set_start_offset_calls{0};
+};
+
 } // namespace
 
 using namespace std::chrono_literals;
@@ -224,6 +235,10 @@ class HousekeeperTest : public testing::Test {
         _l0_metastore.set_max_allowed_start_offset(offset);
     }
 
+    size_t l1_set_start_offset_calls() const {
+        return _l1_metastore.set_start_offset_calls;
+    }
+
     kafka::offset l1_start_offset() {
         auto result = _l1_metastore.get_offsets(_tidp).get();
         if (!result.has_value()) {
@@ -289,7 +304,7 @@ class HousekeeperTest : public testing::Test {
     model::topic_id_partition _tidp{
       model::create_topic_id(), model::partition_id{0}};
     fake_l0_metastore _l0_metastore{_tidp, kafka::offset{0}};
-    cloud_topics::l1::simple_metastore _l1_metastore;
+    counting_metastore _l1_metastore;
     std::unique_ptr<retention_config_impl> _config_impl;
 };
 
@@ -848,3 +863,21 @@ TEST_F(HousekeeperTest, BumpEpochMultipleIdleCycles) {
     EXPECT_EQ(advance_epoch_calls()[0], cloud_topics::cluster_epoch{10});
     EXPECT_EQ(sync_to_next_placeholder_calls(), 1);
 }
+
+TEST_F(HousekeeperTest, SyncStartOffsetSkipsRedundantRPC) {
+    auto housekeeper = make_housekeeper({});
+    add_object({.records = 100, .size = 1_MiB});
+    set_start_offset(kafka::offset{50});
+
+    housekeeper.do_housekeeping().get();
+    EXPECT_EQ(l1_set_start_offset_calls(), 1);
+
+    // No change — should skip the call.
+    housekeeper.do_housekeeping().get();
+    EXPECT_EQ(l1_set_start_offset_calls(), 1);
+
+    // Changed offset — should sync again.
+    set_start_offset(kafka::offset{75});
+    housekeeper.do_housekeeping().get();
+    EXPECT_EQ(l1_set_start_offset_calls(), 2);
+}