Merge pull request #31002 from andrwng/metastore-retry-configs

ct/l1/metastore: add configurations for RPC retries

Author: andrwng

src/v/cloud_topics/level_one/metastore/BUILD

@@ -176,9 +176,15 @@ redpanda_cc_library(
 
 redpanda_cc_library(
     name = "retry",
+    srcs = [
+        "retry.cc",
+    ],
     hdrs = [
         "retry.h",
     ],
+    implementation_deps = [
+        "//src/v/config",
+    ],
     visibility = ["//visibility:public"],
     deps = [
         ":metastore",

src/v/cloud_topics/level_one/metastore/leader_router.cc

@@ -16,6 +16,7 @@
 #include "cluster/metadata_cache.h"
 #include "cluster/partition_leaders_table.h"
 #include "cluster/shard_table.h"
+#include "config/configuration.h"
 #include "hashing/murmur.h"
 #include "model/namespace.h"
 #include "resource_mgmt/cpu_scheduling.h"
@@ -250,13 +251,16 @@ requires requires(
 ss::future<typename req_t::resp_t>
 leader_router::remote_dispatch(req_t request, model::node_id leader_id) {
     using resp_t = req_t::resp_t;
+    auto rpc_timeout
+      = config::shard_local_cfg().cloud_topics_metastore_rpc_timeout_ms();
     auto res = co_await _connection_cache->local()
                  .with_node_client<proto_t>(
                    _self,
                    ss::this_shard_id(),
                    leader_id,
                    rpc_timeout,
-                   [request = std::move(request)](proto_t proto) mutable {
+                   [request = std::move(request),
+                    rpc_timeout](proto_t proto) mutable {
                        return (proto.*Func)(
                          std::move(request),
                          ::rpc::client_opts{

src/v/cloud_topics/level_one/metastore/leader_router.h

@@ -134,8 +134,6 @@ class leader_router : public ss::peering_sharded_service<leader_router> {
     using proto_t = cloud_topics::l1::rpc::impl::l1_rpc_client_protocol;
     using client = cloud_topics::l1::rpc::impl::l1_rpc_client_protocol;
 
-    static constexpr std::chrono::seconds rpc_timeout{5};
-
     // utilities for boiler plate RPC code.
 
     template<auto Func, typename req_t>

src/v/cloud_topics/level_one/metastore/retry.cc

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Licensed as a Redpanda Enterprise file under the Redpanda Community
+ * License (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
+ */
+#include "cloud_topics/level_one/metastore/retry.h"
+
+#include "config/configuration.h"
+
+#include <seastar/core/lowres_clock.hh>
+
+namespace cloud_topics::l1 {
+
+retry_chain_node make_default_metastore_rtc(ss::abort_source& as) {
+    return {
+      as,
+      ss::lowres_clock::now()
+        + config::shard_local_cfg().cloud_topics_metastore_retry_timeout_ms(),
+      default_metastore_retry_backoff};
+}
+
+} // namespace cloud_topics::l1

src/v/cloud_topics/level_one/metastore/retry.h

@@ -24,15 +24,11 @@ namespace cloud_topics::l1 {
 
 using namespace std::chrono_literals;
 
-static constexpr auto default_metastore_retry_timeout = 5s;
 static constexpr auto default_metastore_retry_backoff = 100ms;
 
-inline retry_chain_node make_default_metastore_rtc(ss::abort_source& as) {
-    return {
-      as,
-      ss::lowres_clock::now() + default_metastore_retry_timeout,
-      default_metastore_retry_backoff};
-}
+/// Build a retry_chain_node whose overall deadline is driven by the
+/// cloud_topics_metastore_retry_timeout_ms cluster config property.
+retry_chain_node make_default_metastore_rtc(ss::abort_source& as);
 
 template<typename Func>
 concept metastore_operation = requires(Func f) {

src/v/config/configuration.cc

@@ -5073,6 +5073,22 @@ configuration::configuration()
       "behind and writes are being throttled.",
       {.needs_restart = needs_restart::no, .visibility = visibility::tunable},
       5min)
+  , cloud_topics_metastore_rpc_timeout_ms(
+      *this,
+      "cloud_topics_metastore_rpc_timeout_ms",
+      "Timeout for a single L1 metastore RPC to the metastore partition "
+      "leader. Bounds one attempt; the overall operation may retry until "
+      "cloud_topics_metastore_retry_timeout_ms elapses.",
+      {.needs_restart = needs_restart::no, .visibility = visibility::tunable},
+      30s)
+  , cloud_topics_metastore_retry_timeout_ms(
+      *this,
+      "cloud_topics_metastore_retry_timeout_ms",
+      "Overall deadline for retrying an L1 metastore operation on transport "
+      "errors. To allow more than one attempt, keep this larger than "
+      "cloud_topics_metastore_rpc_timeout_ms.",
+      {.needs_restart = needs_restart::no, .visibility = visibility::tunable},
+      90s)
   , cloud_topics_metastore_block_cache_size(
       *this,
       "cloud_topics_metastore_block_cache_size",

src/v/config/configuration.h

@@ -852,6 +852,8 @@ struct configuration final : public config_store {
       cloud_topics_metastore_replication_timeout_ms;
     property<std::chrono::milliseconds>
       cloud_topics_metastore_lsm_apply_timeout_ms;
+    property<std::chrono::milliseconds> cloud_topics_metastore_rpc_timeout_ms;
+    property<std::chrono::milliseconds> cloud_topics_metastore_retry_timeout_ms;
     bounded_property<size_t> cloud_topics_metastore_block_cache_size;
     bounded_property<size_t> cloud_topics_metastore_write_buffer_size;
     bounded_property<uint32_t> cloud_topics_metastore_max_pre_open_fibers;