Update ScyllaDB version to: 2025.3.0

The test test_tombstone_gc_disabled_on_pending_replica was added when
we fixed (#20788) the potential problem with data resurrection during
file based streaming. The issue was occurring only in Enterprise, but
we added the fix in OSS to limit code divergence. This test was added
together with the fix in OSS with the idea to guard this change in OSS.
The real reproducer and test for this fix was added later, after the
fix was ported into Enterprise.
It is in: test/cluster/test_resurrection.py

Since Enterprise has been merged into OSS, there is no more need to
keep the test test_tombstone_gc_disabled_on_pending_replica. Also,
it is flaky with very low probability of failure, making it difficult
to investigate the cause of failure.

Fixes: #22182

Refs: scylladb/scylladb#25448

Closes scylladb/scylladb#25134

(cherry picked from commit 7ce96345bf)

Closes scylladb/scylladb#25573

.github/scripts/auto-backport.py

@@ -112,10 +112,15 @@ def backport(repo, pr, version, commits, backport_base_branch, is_collaborator):
                     is_draft = True
                     repo_local.git.add(A=True)
                     repo_local.git.cherry_pick('--continue')
-            repo_local.git.push(fork_repo, new_branch_name, force=True)
-            create_pull_request(repo, new_branch_name, backport_base_branch, pr, backport_pr_title, commits,
-                                is_draft, is_collaborator)
-
+            # Check if the branch already exists in the remote fork
+            remote_refs = repo_local.git.ls_remote('--heads', fork_repo, new_branch_name)
+            if not remote_refs:
+                # Branch does not exist, create it with a regular push
+                repo_local.git.push(fork_repo, new_branch_name)
+                create_pull_request(repo, new_branch_name, backport_base_branch, pr, backport_pr_title, commits,
+                                    is_draft, is_collaborator)
+            else:
+                logging.info(f"Remote branch {new_branch_name} already exists in fork. Skipping push.")
         except GitCommandError as e:
             logging.warning(f"GitCommandError: {e}")
 

SCYLLA-VERSION-GEN

@@ -78,7 +78,7 @@ fi
 
 # Default scylla product/version tags
 PRODUCT=scylla
-VERSION=2025.3.0-dev
+VERSION=2025.3.0
 
 if test -f version
 then

alternator/executor.cc

@@ -38,7 +38,6 @@
 #include <optional>
 #include "utils/assert.hh"
 #include "utils/overloaded_functor.hh"
-#include <seastar/json/json_elements.hh>
 #include "collection_mutation.hh"
 #include "schema/schema.hh"
 #include "db/tags/extension.hh"
@@ -121,47 +120,50 @@ static lw_shared_ptr<stats> get_stats_from_schema(service::storage_proxy& sp, co
     }
 }
 
-make_jsonable::make_jsonable(rjson::value&& value)
-    : _value(std::move(value))
-{}
-std::string make_jsonable::to_json() const {
-    return rjson::print(_value);
-}
-
-json::json_return_type make_streamed(rjson::value&& value) {
-    // CMH. json::json_return_type uses std::function, not noncopyable_function.
-    // Need to make a copyable version of value. Gah.
-    auto rs = make_shared<rjson::value>(std::move(value));
-    std::function<future<>(output_stream<char>&&)> func = [rs](output_stream<char>&& os) mutable -> future<> {
-        // move objects to coroutine frame.
-        auto los = std::move(os);
-        auto lrs = std::move(rs);
+executor::body_writer make_streamed(rjson::value&& value) {
+    return [value = std::move(value)](output_stream<char>&& _out) mutable -> future<> {
+        auto out = std::move(_out);
         std::exception_ptr ex;
         try {
-            co_await rjson::print(*lrs, los);
+            co_await rjson::print(value, out);
         } catch (...) {
-            // at this point, we cannot really do anything. HTTP headers and return code are
-            // already written, and quite potentially a portion of the content data.
-            // just log + rethrow. It is probably better the HTTP server closes connection
-            // abruptly or something...
             ex = std::current_exception();
-            elogger.error("Exception during streaming HTTP response: {}", ex);
         }
-        co_await los.close();
-        co_await rjson::destroy_gently(std::move(*lrs));
+        co_await out.close();
+        co_await rjson::destroy_gently(std::move(value));
         if (ex) {
             co_await coroutine::return_exception_ptr(std::move(ex));
         }
-        co_return;
     };
-    return func;
 }
 
-json_string::json_string(std::string&& value)
-    : _value(std::move(value))
-{}
-std::string json_string::to_json() const {
-    return _value;
+// make_streamed_with_extra_array() is variant of make_streamed() above, which
+// builds a streaming response (a function writing to an output stream) from a
+// JSON object (rjson::value) but adds to it at the end an additional array.
+// The extra array is given a separate chunked_vector to avoid putting it
+// inside the rjson::value - because RapidJSON does contiguous allocations for
+// arrays which we want to avoid for potentially long arrays in Query/Scan
+// responses (see #23535).
+// If we ever fix RapidJSON to avoid contiguous allocations for arrays, or
+// replace it entirely (#24458), we can remove this function and the function
+// rjson::print_with_extra_array() which it calls.
+executor::body_writer make_streamed_with_extra_array(rjson::value&& value,
+    std::string array_name, utils::chunked_vector<rjson::value>&& array) {
+    return [value = std::move(value), array_name = std::move(array_name), array = std::move(array)](output_stream<char>&& _out) mutable -> future<> {
+        auto out = std::move(_out);
+        std::exception_ptr ex;
+        try {
+            co_await rjson::print_with_extra_array(value, array_name, array, out);
+        } catch (...) {
+            ex = std::current_exception();
+        }
+        co_await out.close();
+        co_await rjson::destroy_gently(std::move(value));
+        // TODO: can/should we also destroy the array gently?
+        if (ex) {
+            co_await coroutine::return_exception_ptr(std::move(ex));
+        }
+    };
 }
 
 // This function throws api_error::validation if input value is not an object.
@@ -764,7 +766,7 @@ future<executor::request_return_type> executor::describe_table(client_state& cli
     rjson::value response = rjson::empty_object();
     rjson::add(response, "Table", std::move(table_description));
     elogger.trace("returning {}", response);
-    co_return make_jsonable(std::move(response));
+    co_return rjson::print(std::move(response));
 }
 
 // Check CQL's Role-Based Access Control (RBAC) permission_to_check (MODIFY,
@@ -881,7 +883,7 @@ future<executor::request_return_type> executor::delete_table(client_state& clien
     rjson::value response = rjson::empty_object();
     rjson::add(response, "TableDescription", std::move(table_description));
     elogger.trace("returning {}", response);
-    co_return make_jsonable(std::move(response));
+    co_return rjson::print(std::move(response));
 }
 
 static data_type parse_key_type(std::string_view type) {
@@ -1165,7 +1167,7 @@ future<executor::request_return_type> executor::tag_resource(client_state& clien
     co_await db::modify_tags(_mm, schema->ks_name(), schema->cf_name(), [tags](std::map<sstring, sstring>& tags_map) {
         update_tags_map(*tags, tags_map, update_tags_action::add_tags);
     });
-    co_return json_string("");
+    co_return ""; // empty response
 }
 
 future<executor::request_return_type> executor::untag_resource(client_state& client_state, service_permit permit, rjson::value request) {
@@ -1186,7 +1188,7 @@ future<executor::request_return_type> executor::untag_resource(client_state& cli
     co_await db::modify_tags(_mm, schema->ks_name(), schema->cf_name(), [tags](std::map<sstring, sstring>& tags_map) {
         update_tags_map(*tags, tags_map, update_tags_action::delete_tags);
     });
-    co_return json_string("");
+    co_return ""; // empty response
 }
 
 future<executor::request_return_type> executor::list_tags_of_resource(client_state& client_state, service_permit permit, rjson::value request) {
@@ -1212,7 +1214,7 @@ future<executor::request_return_type> executor::list_tags_of_resource(client_sta
         rjson::push_back(tags, std::move(new_entry));
     }
 
-    return make_ready_future<executor::request_return_type>(make_jsonable(std::move(ret)));
+    return make_ready_future<executor::request_return_type>(rjson::print(std::move(ret)));
 }
 
 struct billing_mode_type {
@@ -1674,7 +1676,7 @@ static future<executor::request_return_type> create_table_on_shard0(service::cli
     rjson::value status = rjson::empty_object();
     executor::supplement_table_info(request, *schema, sp);
     rjson::add(status, "TableDescription", std::move(request));
-    co_return make_jsonable(std::move(status));
+    co_return rjson::print(std::move(status));
 }
 
 future<executor::request_return_type> executor::create_table(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
@@ -1951,7 +1953,7 @@ future<executor::request_return_type> executor::update_table(client_state& clien
         rjson::value status = rjson::empty_object();
         supplement_table_info(request, *schema, p.local());
         rjson::add(status, "TableDescription", std::move(request));
-        co_return make_jsonable(std::move(status));
+        co_return rjson::print(std::move(status));
     });
 }
 
@@ -2417,7 +2419,7 @@ static future<executor::request_return_type> rmw_operation_return(rjson::value&&
     if (!attributes.IsNull()) {
         rjson::add(ret, "Attributes", std::move(attributes));
     }
-    return make_ready_future<executor::request_return_type>(make_jsonable(std::move(ret)));
+    return make_ready_future<executor::request_return_type>(rjson::print(std::move(ret)));
 }
 
 static future<std::unique_ptr<rjson::value>> get_previous_item(
@@ -3009,7 +3011,7 @@ future<executor::request_return_type> executor::batch_write_item(client_state& c
         rjson::add(ret, "ConsumedCapacity", std::move(consumed_capacity));
     }
     _stats.api_operations.batch_write_item_latency.mark(std::chrono::steady_clock::now() - start_time);
-    co_return make_jsonable(std::move(ret));
+    co_return rjson::print(std::move(ret));
 }
 
 static const std::string_view get_item_type_string(const rjson::value& v) {
@@ -4249,18 +4251,17 @@ future<executor::request_return_type> executor::get_item(client_state& client_st
     verify_all_are_used(expression_attribute_names, used_attribute_names, "ExpressionAttributeNames", "GetItem");
     rcu_consumed_capacity_counter add_capacity(request, cl == db::consistency_level::LOCAL_QUORUM);
     co_await verify_permission(_enforce_authorization, client_state, schema, auth::permission::SELECT);
-    co_return co_await _proxy.query(schema, std::move(command), std::move(partition_ranges), cl,
-            service::storage_proxy::coordinator_query_options(executor::default_timeout(), std::move(permit), client_state, trace_state)).then(
-            [per_table_stats, this, schema, partition_slice = std::move(partition_slice), selection = std::move(selection), attrs_to_get = std::move(attrs_to_get), start_time = std::move(start_time), add_capacity=std::move(add_capacity)] (service::storage_proxy::coordinator_query_result qr) mutable {
-
-        per_table_stats->api_operations.get_item_latency.mark(std::chrono::steady_clock::now() - start_time);
-        _stats.api_operations.get_item_latency.mark(std::chrono::steady_clock::now() - start_time);
-        uint64_t rcu_half_units = 0;
-        auto res = make_ready_future<executor::request_return_type>(make_jsonable(describe_item(schema, partition_slice, *selection, *qr.query_result, std::move(attrs_to_get), add_capacity, rcu_half_units)));
-        per_table_stats->rcu_half_units_total += rcu_half_units;
-        _stats.rcu_half_units_total += rcu_half_units;
-        return res;
-    });
+    service::storage_proxy::coordinator_query_result qr =
+        co_await _proxy.query(
+            schema, std::move(command), std::move(partition_ranges), cl,
+            service::storage_proxy::coordinator_query_options(executor::default_timeout(), std::move(permit), client_state, trace_state));
+    per_table_stats->api_operations.get_item_latency.mark(std::chrono::steady_clock::now() - start_time);
+    _stats.api_operations.get_item_latency.mark(std::chrono::steady_clock::now() - start_time);
+    uint64_t rcu_half_units = 0;
+    rjson::value res = describe_item(schema, partition_slice, *selection, *qr.query_result, std::move(attrs_to_get), add_capacity, rcu_half_units);
+    per_table_stats->rcu_half_units_total += rcu_half_units;
+    _stats.rcu_half_units_total += rcu_half_units;
+    co_return rjson::print(std::move(res));
 }
 
 static void check_big_object(const rjson::value& val, int& size_left);
@@ -4505,7 +4506,7 @@ future<executor::request_return_type> executor::batch_get_item(client_state& cli
     if (is_big(response)) {
         co_return make_streamed(std::move(response));
     } else {
-        co_return make_jsonable(std::move(response));
+        co_return rjson::print(std::move(response));
     }
 }
 
@@ -4649,7 +4650,11 @@ class describe_items_visitor {
     const filter& _filter;
     typename columns_t::const_iterator _column_it;
     rjson::value _item;
-    rjson::value _items;
+    // _items is a chunked_vector<rjson::value> instead of a RapidJson array
+    // (rjson::value) because unfortunately RapidJson arrays are stored
+    // contiguously in memory, and cause large allocations when a Query/Scan
+    // returns a long list of short items (issue #23535).
+    utils::chunked_vector<rjson::value> _items;
     size_t _scanned_count;
 
 public:
@@ -4659,7 +4664,6 @@ public:
             , _filter(filter)
             , _column_it(columns.begin())
             , _item(rjson::empty_object())
-            , _items(rjson::empty_array())
             , _scanned_count(0)
     {
         // _filter.check() may need additional attributes not listed in
@@ -4738,13 +4742,13 @@ public:
                 rjson::remove_member(_item, attr);
             }
 
-            rjson::push_back(_items, std::move(_item));
+            _items.push_back(std::move(_item));
         }
         _item = rjson::empty_object();
         ++_scanned_count;
     }
 
-    rjson::value get_items() && {
+    utils::chunked_vector<rjson::value> get_items() && {
         return std::move(_items);
     }
 
@@ -4753,13 +4757,25 @@ public:
     }
 };
 
-static future<std::tuple<rjson::value, size_t>> describe_items(const cql3::selection::selection& selection, std::unique_ptr<cql3::result_set> result_set, std::optional<attrs_to_get>&& attrs_to_get, filter&& filter) {
+// describe_items() returns a JSON object that includes members "Count"
+// and "ScannedCount", but *not* "Items" - that is returned separately
+// as a chunked_vector to avoid large contiguous allocations which
+// RapidJSON does of its array. The caller should add "Items" to the
+// returned JSON object if needed, or print it separately.
+// The returned chunked_vector (the items) is std::optional<>, because
+// the user may have requested only to count items, and not return any
+// items - which is different from returning an empty list of items.
+static future<std::tuple<rjson::value, std::optional<utils::chunked_vector<rjson::value>>, size_t>> describe_items(
+        const cql3::selection::selection& selection,
+        std::unique_ptr<cql3::result_set> result_set,
+        std::optional<attrs_to_get>&& attrs_to_get,
+        filter&& filter) {
     describe_items_visitor visitor(selection.get_columns(), attrs_to_get, filter);
     co_await result_set->visit_gently(visitor);
     auto scanned_count = visitor.get_scanned_count();
-    rjson::value items = std::move(visitor).get_items();
+    utils::chunked_vector<rjson::value> items = std::move(visitor).get_items();
     rjson::value items_descr = rjson::empty_object();
-    auto size = items.Size();
+    auto size = items.size();
     rjson::add(items_descr, "Count", rjson::value(size));
     rjson::add(items_descr, "ScannedCount", rjson::value(scanned_count));
     // If attrs_to_get && attrs_to_get->empty(), this means the user asked not
@@ -4769,10 +4785,11 @@ static future<std::tuple<rjson::value, size_t>> describe_items(const cql3::selec
     // In that case, we currently build a list of empty items and here drop
     // it. We could just count the items and not bother with the empty items.
     // (However, remember that when we do have a filter, we need the items).
+    std::optional<utils::chunked_vector<rjson::value>> opt_items;
     if (!attrs_to_get || !attrs_to_get->empty()) {
-        rjson::add(items_descr, "Items", std::move(items));
+        opt_items = std::move(items);
     }
-    co_return std::tuple<rjson::value, size_t>{std::move(items_descr), size};
+    co_return std::tuple(std::move(items_descr), std::move(opt_items), size);
 }
 
 static rjson::value encode_paging_state(const schema& schema, const service::pager::paging_state& paging_state) {
@@ -4810,6 +4827,12 @@ static rjson::value encode_paging_state(const schema& schema, const service::pag
     return last_evaluated_key;
 }
 
+// RapidJSON allocates arrays contiguously in memory, so we want to avoid
+// returning a large number of items as a single rapidjson array, and use
+// a chunked_vector instead. The following constant is an arbitrary cutoff
+// point for when to switch from a rapidjson array to a chunked_vector.
+static constexpr int max_items_for_rapidjson_array = 256;
+
 static future<executor::request_return_type> do_query(service::storage_proxy& proxy,
         schema_ptr table_schema,
         const rjson::value* exclusive_start_key,
@@ -4882,19 +4905,35 @@ static future<executor::request_return_type> do_query(service::storage_proxy& pr
     }
     auto paging_state = rs->get_metadata().paging_state();
     bool has_filter = filter;
-    auto [items, size] = co_await describe_items(*selection, std::move(rs), std::move(attrs_to_get), std::move(filter));
+    auto [items_descr, opt_items, size] = co_await describe_items(*selection, std::move(rs), std::move(attrs_to_get), std::move(filter));
     if (paging_state) {
-        rjson::add(items, "LastEvaluatedKey", encode_paging_state(*table_schema, *paging_state));
+        rjson::add(items_descr, "LastEvaluatedKey", encode_paging_state(*table_schema, *paging_state));
     }
     if (has_filter){
         cql_stats.filtered_rows_read_total += p->stats().rows_read_total;
         // update our "filtered_row_matched_total" for all the rows matched, despited the filter
         cql_stats.filtered_rows_matched_total += size;
     }
-    if (is_big(items)) {
-        co_return executor::request_return_type(make_streamed(std::move(items)));
+    if (opt_items) {
+        if (opt_items->size() >= max_items_for_rapidjson_array) {
+            // There are many items, better print the JSON and the array of
+            // items (opt_items) separately to avoid RapidJSON's contiguous
+            // allocation of arrays.
+            co_return make_streamed_with_extra_array(std::move(items_descr), "Items", std::move(*opt_items));
+        }
+        // There aren't many items in the chunked vector opt_items,
+        // let's just insert them into the JSON object and print the
+        // full JSON normally.
+        rjson::value items_json = rjson::empty_array();
+        for (auto& item : *opt_items) {
+            rjson::push_back(items_json, std::move(item));
+        }
+        rjson::add(items_descr, "Items", std::move(items_json));
     }
-    co_return executor::request_return_type(make_jsonable(std::move(items)));
+    if (is_big(items_descr)) {
+        co_return make_streamed(std::move(items_descr));
+    }
+    co_return rjson::print(std::move(items_descr));
 }
 
 static dht::token token_for_segment(int segment, int total_segments) {
@@ -5489,7 +5528,7 @@ future<executor::request_return_type> executor::list_tables(client_state& client
     std::string exclusive_start = exclusive_start_json ? exclusive_start_json->GetString() : "";
     int limit = limit_json ? limit_json->GetInt() : 100;
     if (limit < 1 || limit > 100) {
-        return make_ready_future<request_return_type>(api_error::validation("Limit must be greater than 0 and no greater than 100"));
+        co_return api_error::validation("Limit must be greater than 0 and no greater than 100");
     }
 
     auto tables = _proxy.data_dictionary().get_tables(); // hold on to temporary, table_names isn't a container, it's a view
@@ -5531,7 +5570,7 @@ future<executor::request_return_type> executor::list_tables(client_state& client
         rjson::add(response, "LastEvaluatedTableName", rjson::copy(last_table_name));
     }
 
-    return make_ready_future<executor::request_return_type>(make_jsonable(std::move(response)));
+    co_return rjson::print(std::move(response));
 }
 
 future<executor::request_return_type> executor::describe_endpoints(client_state& client_state, service_permit permit, rjson::value request, std::string host_header) {
@@ -5542,8 +5581,8 @@ future<executor::request_return_type> executor::describe_endpoints(client_state&
     if (!override.empty()) {
         if (override == "disabled") {
             _stats.unsupported_operations++;
-            return make_ready_future<request_return_type>(api_error::unknown_operation(
-                "DescribeEndpoints disabled by configuration (alternator_describe_endpoints=disabled)"));
+            co_return api_error::unknown_operation(
+                "DescribeEndpoints disabled by configuration (alternator_describe_endpoints=disabled)");
         }
         host_header = std::move(override);
     }
@@ -5555,13 +5594,13 @@ future<executor::request_return_type> executor::describe_endpoints(client_state&
     // A "Host:" header includes both host name and port, exactly what we need
     // to return.
     if (host_header.empty()) {
-        return make_ready_future<request_return_type>(api_error::validation("DescribeEndpoints needs a 'Host:' header in request"));
+        co_return api_error::validation("DescribeEndpoints needs a 'Host:' header in request");
     }
     rjson::add(response, "Endpoints", rjson::empty_array());
     rjson::push_back(response["Endpoints"], rjson::empty_object());
     rjson::add(response["Endpoints"][0], "Address", rjson::from_string(host_header));
     rjson::add(response["Endpoints"][0], "CachePeriodInMinutes", rjson::value(1440));
-    return make_ready_future<executor::request_return_type>(make_jsonable(std::move(response)));
+    co_return rjson::print(std::move(response));
 }
 
 static std::map<sstring, sstring> get_network_topology_options(service::storage_proxy& sp, gms::gossiper& gossiper, int rf) {
@@ -5596,7 +5635,7 @@ future<executor::request_return_type> executor::describe_continuous_backups(clie
     rjson::add(desc, "PointInTimeRecoveryDescription", std::move(pitr));
     rjson::value response = rjson::empty_object();
     rjson::add(response, "ContinuousBackupsDescription", std::move(desc));
-    co_return make_jsonable(std::move(response));
+    co_return rjson::print(std::move(response));
 }
 
 // Create the metadata for the keyspace in which we put the alternator

alternator/executor.hh

@@ -10,8 +10,8 @@
 
 #include <seastar/core/future.hh>
 #include "seastarx.hh"
-#include <seastar/json/json_elements.hh>
 #include <seastar/core/sharded.hh>
+#include <seastar/util/noncopyable_function.hh>
 
 #include "service/migration_manager.hh"
 #include "service/client_state.hh"
@@ -58,29 +58,6 @@ namespace alternator {
 
 class rmw_operation;
 
-struct make_jsonable : public json::jsonable {
-    rjson::value _value;
-public:
-    explicit make_jsonable(rjson::value&& value);
-    std::string to_json() const override;
-};
-
-/**
- * Make return type for serializing the object "streamed",
- * i.e. direct to HTTP output stream. Note: only useful for
- * (very) large objects as there are overhead issues with this
- * as well, but for massive lists of return objects this can
- * help avoid large allocations/many re-allocs
- */
-json::json_return_type make_streamed(rjson::value&&);
-
-struct json_string : public json::jsonable {
-    std::string _value;
-public:
-    explicit json_string(std::string&& value);
-    std::string to_json() const override;
-};
-
 namespace parsed {
 class path;
 };
@@ -169,7 +146,19 @@ class executor : public peering_sharded_service<executor> {
 
 public:
     using client_state = service::client_state;
-    using request_return_type = std::variant<json::json_return_type, api_error>;
+    // request_return_type is the return type of the executor methods, which
+    // can be one of:
+    // 1. A string, which is the response body for the request.
+    // 2. A body_writer, an asynchronous function (returning future<>) that
+    //    takes an output_stream and writes the response body into it.
+    // 3. An api_error, which is an error response that should be returned to
+    //    the client.
+    // The body_writer is used for streaming responses, where the response body
+    // is written in chunks to the output_stream. This allows for efficient
+    // handling of large responses without needing to allocate a large buffer
+    // in memory.
+    using body_writer = noncopyable_function<future<>(output_stream<char>&&)>;
+    using request_return_type = std::variant<std::string, body_writer, api_error>;
     stats _stats;
     // The metric_groups object holds this stat object's metrics registered
     // as long as the stats object is alive.
@@ -275,4 +264,13 @@ bool is_big(const rjson::value& val, int big_size = 100'000);
 // appropriate user-readable api_error::access_denied is thrown.
 future<> verify_permission(bool enforce_authorization, const service::client_state&, const schema_ptr&, auth::permission);
 
+/**
+ * Make return type for serializing the object "streamed",
+ * i.e. direct to HTTP output stream. Note: only useful for
+ * (very) large objects as there are overhead issues with this
+ * as well, but for massive lists of return objects this can
+ * help avoid large allocations/many re-allocs
+ */
+executor::body_writer make_streamed(rjson::value&&);
+
 }

alternator/server.cc

@@ -13,7 +13,6 @@
 #include <seastar/http/function_handlers.hh>
 #include <seastar/http/short_streams.hh>
 #include <seastar/core/coroutine.hh>
-#include <seastar/json/json_elements.hh>
 #include <seastar/util/defer.hh>
 #include <seastar/util/short_streams.hh>
 #include "seastarx.hh"
@@ -124,22 +123,22 @@ public:
              }
              auto res = resf.get();
              std::visit(overloaded_functor {
-                 [&] (const json::json_return_type& json_return_value) {
-                     slogger.trace("api_handler success case");
-                     if (json_return_value._body_writer) {
-                         // Unfortunately, write_body() forces us to choose
-                         // from a fixed and irrelevant list of "mime-types"
-                         // at this point. But we'll override it with the
-                         // one (application/x-amz-json-1.0) below.
-                         rep->write_body("json", std::move(json_return_value._body_writer));
-                     } else {
-                         rep->_content += json_return_value._res;
-                     }
-                 },
-                 [&] (const api_error& err) {
-                     generate_error_reply(*rep, err);
-                 }
-             }, res);
+                [&] (std::string&& str) {
+                    // Note that despite the move, there is a copy here -
+                    // as str is std::string and rep->_content is sstring.
+                    rep->_content = std::move(str);
+                },
+                [&] (executor::body_writer&& body_writer) {
+                    // Unfortunately, write_body() forces us to choose
+                    // from a fixed and irrelevant list of "mime-types"
+                    // at this point. But we'll override it with the
+                    // correct one (application/x-amz-json-1.0) below.
+                    rep->write_body("json", std::move(body_writer));
+                },
+                [&] (const api_error& err) {
+                    generate_error_reply(*rep, err);
+                }
+             }, std::move(res));
 
              return make_ready_future<std::unique_ptr<reply>>(std::move(rep));
          });

alternator/streams.cc

@@ -217,7 +217,7 @@ future<alternator::executor::request_return_type> alternator::executor::list_str
         rjson::add(ret, "LastEvaluatedStreamArn", *last);
     }
 
-    return make_ready_future<executor::request_return_type>(make_jsonable(std::move(ret)));
+    return make_ready_future<executor::request_return_type>(rjson::print(std::move(ret)));
 }
 
 struct shard_id {
@@ -491,7 +491,7 @@ future<executor::request_return_type> executor::describe_stream(client_state& cl
 
     if (!opts.enabled()) {
         rjson::add(ret, "StreamDescription", std::move(stream_desc));
-        return make_ready_future<executor::request_return_type>(make_jsonable(std::move(ret)));
+        return make_ready_future<executor::request_return_type>(rjson::print(std::move(ret)));
     }
 
     // TODO: label
@@ -617,7 +617,7 @@ future<executor::request_return_type> executor::describe_stream(client_state& cl
         rjson::add(stream_desc, "Shards", std::move(shards));
         rjson::add(ret, "StreamDescription", std::move(stream_desc));
             
-        return make_ready_future<executor::request_return_type>(make_jsonable(std::move(ret)));
+        return make_ready_future<executor::request_return_type>(rjson::print(std::move(ret)));
     });
 }
 
@@ -770,7 +770,7 @@ future<executor::request_return_type> executor::get_shard_iterator(client_state&
     auto ret = rjson::empty_object();
     rjson::add(ret, "ShardIterator", iter);
 
-    return make_ready_future<executor::request_return_type>(make_jsonable(std::move(ret)));
+    return make_ready_future<executor::request_return_type>(rjson::print(std::move(ret)));
 }
 
 struct event_id {
@@ -1021,7 +1021,7 @@ future<executor::request_return_type> executor::get_records(client_state& client
             // will notice end end of shard and not return NextShardIterator.
             rjson::add(ret, "NextShardIterator", next_iter);
             _stats.api_operations.get_records_latency.mark(std::chrono::steady_clock::now() - start_time);
-            return make_ready_future<executor::request_return_type>(make_jsonable(std::move(ret)));
+            return make_ready_future<executor::request_return_type>(rjson::print(std::move(ret)));
         }
 
         // ugh. figure out if we are and end-of-shard
@@ -1047,7 +1047,7 @@ future<executor::request_return_type> executor::get_records(client_state& client
             if (is_big(ret)) {
                 return make_ready_future<executor::request_return_type>(make_streamed(std::move(ret)));
             }
-            return make_ready_future<executor::request_return_type>(make_jsonable(std::move(ret)));
+            return make_ready_future<executor::request_return_type>(rjson::print(std::move(ret)));
         });
     });
 }

alternator/ttl.cc

@@ -118,7 +118,7 @@ future<executor::request_return_type> executor::update_time_to_live(client_state
     // basically identical to the request's
     rjson::value response = rjson::empty_object();
     rjson::add(response, "TimeToLiveSpecification", std::move(*spec));
-    co_return make_jsonable(std::move(response));
+    co_return rjson::print(std::move(response));
 }
 
 future<executor::request_return_type> executor::describe_time_to_live(client_state& client_state, service_permit permit, rjson::value request) {
@@ -135,7 +135,7 @@ future<executor::request_return_type> executor::describe_time_to_live(client_sta
     }
     rjson::value response = rjson::empty_object();
     rjson::add(response, "TimeToLiveDescription", std::move(desc));
-    co_return make_jsonable(std::move(response));
+    co_return rjson::print(std::move(response));
 }
 
 // expiration_service is a sharded service responsible for cleaning up expired

api/api-doc/storage_service.json

@@ -3161,6 +3161,22 @@
                ]
             }
          ]
+      },
+      {
+         "path":"/storage_service/raft_topology/cmd_rpc_status",
+         "operations":[
+            {
+               "method":"GET",
+               "summary":"Get information about currently running topology cmd rpc",
+               "type":"string",
+               "nickname":"raft_topology_get_cmd_status",
+               "produces":[
+                  "application/json"
+               ],
+               "parameters":[
+               ]
+            }
+         ]
       }
    ],
    "models":{

api/storage_service.cc

@@ -749,13 +749,7 @@ rest_force_compaction(http_context& ctx, std::unique_ptr<http::request> req) {
             fmopt = flush_mode::skip;
         }
         auto task = co_await compaction_module.make_and_start_task<global_major_compaction_task_impl>({}, db, fmopt, consider_only_existing_data);
-        try {
-            co_await task->done();
-        } catch (...) {
-            apilog.error("force_compaction failed: {}", std::current_exception());
-            throw;
-        }
-
+        co_await task->done();
         co_return json_void();
 }
 
@@ -774,13 +768,7 @@ rest_force_keyspace_compaction(http_context& ctx, std::unique_ptr<http::request>
             fmopt = flush_mode::skip;
         }
         auto task = co_await compaction_module.make_and_start_task<major_keyspace_compaction_task_impl>({}, std::move(keyspace), tasks::task_id::create_null_id(), db, table_infos, fmopt, consider_only_existing_data);
-        try {
-            co_await task->done();
-        } catch (...) {
-            apilog.error("force_keyspace_compaction: keyspace={} tables={} failed: {}", task->get_status().keyspace, table_infos, std::current_exception());
-            throw;
-        }
-
+        co_await task->done();
         co_return json_void();
 }
 
@@ -805,13 +793,7 @@ rest_force_keyspace_cleanup(http_context& ctx, sharded<service::storage_service>
         auto& compaction_module = db.local().get_compaction_manager().get_task_manager_module();
         auto task = co_await compaction_module.make_and_start_task<cleanup_keyspace_compaction_task_impl>(
             {}, std::move(keyspace), db, table_infos, flush_mode::all_tables, tasks::is_user_task::yes);
-        try {
-            co_await task->done();
-        } catch (...) {
-            apilog.error("force_keyspace_cleanup: keyspace={} tables={} failed: {}", task->get_status().keyspace, table_infos, std::current_exception());
-            throw;
-        }
-
+        co_await task->done();
         co_return json::json_return_type(0);
 }
 
@@ -833,12 +815,7 @@ rest_cleanup_all(http_context& ctx, sharded<service::storage_service>& ss, std::
         auto& db = ctx.db;
         auto& compaction_module = db.local().get_compaction_manager().get_task_manager_module();
         auto task = co_await compaction_module.make_and_start_task<global_cleanup_compaction_task_impl>({}, db);
-        try {
-            co_await task->done();
-        } catch (...) {
-            apilog.error("cleanup_all failed: {}", std::current_exception());
-            throw;
-        }
+        co_await task->done();
         co_return json::json_return_type(0);
 }
 
@@ -850,13 +827,7 @@ rest_perform_keyspace_offstrategy_compaction(http_context& ctx, std::unique_ptr<
         bool res = false;
         auto& compaction_module = ctx.db.local().get_compaction_manager().get_task_manager_module();
         auto task = co_await compaction_module.make_and_start_task<offstrategy_keyspace_compaction_task_impl>({}, std::move(keyspace), ctx.db, table_infos, &res);
-        try {
-            co_await task->done();
-        } catch (...) {
-            apilog.error("perform_keyspace_offstrategy_compaction: keyspace={} tables={} failed: {}", task->get_status().keyspace, table_infos, std::current_exception());
-            throw;
-        }
-
+        co_await task->done();
         co_return json::json_return_type(res);
 }
 
@@ -871,13 +842,7 @@ rest_upgrade_sstables(http_context& ctx, std::unique_ptr<http::request> req) {
 
         auto& compaction_module = db.local().get_compaction_manager().get_task_manager_module();
         auto task = co_await compaction_module.make_and_start_task<upgrade_sstables_compaction_task_impl>({}, std::move(keyspace), db, table_infos, exclude_current_version);
-        try {
-            co_await task->done();
-        } catch (...) {
-            apilog.error("upgrade_sstables: keyspace={} tables={} failed: {}", keyspace, table_infos, std::current_exception());
-            throw;
-        }
-
+        co_await task->done();
         co_return json::json_return_type(0);
 }
 
@@ -1670,6 +1635,18 @@ rest_raft_topology_upgrade_status(sharded<service::storage_service>& ss, std::un
         co_return sstring(format("{}", ustate));
 }
 
+static
+future<json::json_return_type>
+rest_raft_topology_get_cmd_status(sharded<service::storage_service>& ss, std::unique_ptr<http::request> req) {
+        const auto status = co_await ss.invoke_on(0, [] (auto& ss) {
+            return ss.get_topology_cmd_status();
+        });
+        if (status.active_dst.empty()) {
+            co_return sstring("none");
+        }
+        co_return sstring(fmt::format("{}[{}]: {}", status.current, status.index, fmt::join(status.active_dst, ",")));
+}
+
 static
 future<json::json_return_type>
 rest_move_tablet(http_context& ctx, sharded<service::storage_service>& ss, std::unique_ptr<http::request> req) {
@@ -1902,6 +1879,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
     ss::reload_raft_topology_state.set(r, rest_bind(rest_reload_raft_topology_state, ss, group0_client));
     ss::upgrade_to_raft_topology.set(r, rest_bind(rest_upgrade_to_raft_topology, ss));
     ss::raft_topology_upgrade_status.set(r, rest_bind(rest_raft_topology_upgrade_status, ss));
+    ss::raft_topology_get_cmd_status.set(r, rest_bind(rest_raft_topology_get_cmd_status, ss));
     ss::move_tablet.set(r, rest_bind(rest_move_tablet, ctx, ss));
     ss::add_tablet_replica.set(r, rest_bind(rest_add_tablet_replica, ctx, ss));
     ss::del_tablet_replica.set(r, rest_bind(rest_del_tablet_replica, ctx, ss));
@@ -1983,6 +1961,7 @@ void unset_storage_service(http_context& ctx, routes& r) {
     ss::reload_raft_topology_state.unset(r);
     ss::upgrade_to_raft_topology.unset(r);
     ss::raft_topology_upgrade_status.unset(r);
+    ss::raft_topology_get_cmd_status.unset(r);
     ss::move_tablet.unset(r);
     ss::add_tablet_replica.unset(r);
     ss::del_tablet_replica.unset(r);

auth/password_authenticator.cc

@@ -227,7 +227,9 @@ future<> password_authenticator::start() {
                 utils::get_local_injector().inject("password_authenticator_start_pause", utils::wait_for_message(5min)).get();
                 if (!legacy_mode(_qp)) {
                     maybe_create_default_password_with_retries().get();
-                    _superuser_created_promise.set_value();
+                    if (!_superuser_created_promise.available()) {
+                        _superuser_created_promise.set_value();
+                    }
                 }
             });
         });

auth/standard_role_manager.cc

@@ -9,6 +9,7 @@
 #include "auth/standard_role_manager.hh"
 
 #include <optional>
+#include <stdexcept>
 #include <unordered_set>
 #include <vector>
 
@@ -28,6 +29,7 @@
 #include "cql3/util.hh"
 #include "db/consistency_level_type.hh"
 #include "exceptions/exceptions.hh"
+#include "utils/error_injection.hh"
 #include "utils/log.hh"
 #include <seastar/core/loop.hh>
 #include <seastar/coroutine/maybe_yield.hh>
@@ -321,7 +323,9 @@ future<> standard_role_manager::start() {
             }
             if (!legacy) {
                 co_await maybe_create_default_role_with_retries();
-                _superuser_created_promise.set_value();
+                if (!_superuser_created_promise.available()) {
+                    _superuser_created_promise.set_value();
+                }
             }
         };
 
@@ -671,6 +675,12 @@ future<role_set> standard_role_manager::query_all() {
     // To avoid many copies of a view.
     static const auto role_col_name_string = sstring(meta::roles_table::role_col_name);
 
+    if (utils::get_local_injector().enter("standard_role_manager_fail_legacy_query")) {
+        if (legacy_mode(_qp)) {
+            throw std::runtime_error("standard_role_manager::query_all: failed due to error injection");
+        }
+    }
+
     const auto results = co_await _qp.execute_internal(
             query,
             db::consistency_level::QUORUM,

cdc/log.cc

@@ -960,8 +960,12 @@ public:
     // Given a reference to such a column from the base schema, this function sets the corresponding column
     // in the log to the given value for the given row.
     void set_value(const clustering_key& log_ck, const column_definition& base_cdef, const managed_bytes_view& value) {
-        auto& log_cdef = *_log_schema.get_column_definition(log_data_column_name_bytes(base_cdef.name()));
-        _log_mut.set_cell(log_ck, log_cdef, atomic_cell::make_live(*base_cdef.type, _ts, value, _ttl));
+        auto log_cdef_ptr = _log_schema.get_column_definition(log_data_column_name_bytes(base_cdef.name()));
+        if (!log_cdef_ptr) {
+            throw exceptions::invalid_request_exception(format("CDC log schema for {}.{} does not have base column {}",
+                _log_schema.ks_name(), _log_schema.cf_name(), base_cdef.name_as_text()));
+        }
+        _log_mut.set_cell(log_ck, *log_cdef_ptr, atomic_cell::make_live(*base_cdef.type, _ts, value, _ttl));
     }
 
     // Each regular and static column in the base schema has a corresponding column in the log schema
@@ -969,7 +973,13 @@ public:
     // Given a reference to such a column from the base schema, this function sets the corresponding column
     // in the log to `true` for the given row. If not called, the column will be `null`.
     void set_deleted(const clustering_key& log_ck, const column_definition& base_cdef) {
-        _log_mut.set_cell(log_ck, log_data_column_deleted_name_bytes(base_cdef.name()), data_value(true), _ts, _ttl);
+        auto log_cdef_ptr = _log_schema.get_column_definition(log_data_column_deleted_name_bytes(base_cdef.name()));
+        if (!log_cdef_ptr) {
+            throw exceptions::invalid_request_exception(format("CDC log schema for {}.{} does not have base column {}",
+                _log_schema.ks_name(), _log_schema.cf_name(), base_cdef.name_as_text()));
+        }
+        auto& log_cdef = *log_cdef_ptr;
+        _log_mut.set_cell(log_ck, *log_cdef_ptr, atomic_cell::make_live(*log_cdef.type, _ts, log_cdef.type->decompose(true), _ttl));
     }
 
     // Each regular and static non-atomic column in the base schema has a corresponding column in the log schema
@@ -978,7 +988,12 @@ public:
     // Given a reference to such a column from the base schema, this function sets the corresponding column
     // in the log to the given set of keys for the given row.
     void set_deleted_elements(const clustering_key& log_ck, const column_definition& base_cdef, const managed_bytes& deleted_elements) {
-        auto& log_cdef = *_log_schema.get_column_definition(log_data_column_deleted_elements_name_bytes(base_cdef.name()));
+        auto log_cdef_ptr = _log_schema.get_column_definition(log_data_column_deleted_elements_name_bytes(base_cdef.name()));
+        if (!log_cdef_ptr) {
+            throw exceptions::invalid_request_exception(format("CDC log schema for {}.{} does not have base column {}",
+                _log_schema.ks_name(), _log_schema.cf_name(), base_cdef.name_as_text()));
+        }
+        auto& log_cdef = *log_cdef_ptr;
         _log_mut.set_cell(log_ck, log_cdef, atomic_cell::make_live(*log_cdef.type, _ts, deleted_elements, _ttl));
     }
 
@@ -1865,5 +1880,10 @@ bool cdc::cdc_service::needs_cdc_augmentation(const std::vector<mutation>& mutat
 
 future<std::tuple<std::vector<mutation>, lw_shared_ptr<cdc::operation_result_tracker>>>
 cdc::cdc_service::augment_mutation_call(lowres_clock::time_point timeout, std::vector<mutation>&& mutations, tracing::trace_state_ptr tr_state, db::consistency_level write_cl) {
+    if (utils::get_local_injector().enter("sleep_before_cdc_augmentation")) {
+        return seastar::sleep(std::chrono::milliseconds(100)).then([this, timeout, mutations = std::move(mutations), tr_state = std::move(tr_state), write_cl] () mutable {
+            return _impl->augment_mutation_call(timeout, std::move(mutations), std::move(tr_state), write_cl);
+        });
+    }
     return _impl->augment_mutation_call(timeout, std::move(mutations), std::move(tr_state), write_cl);
 }

conf/scylla.yaml

@@ -855,3 +855,18 @@ rf_rack_valid_keyspaces: false
 # Maximum number of items in single BatchWriteItem command. Default is 100.
 # Note: DynamoDB has a hard-coded limit of 25.
 # alternator_max_items_in_batch_write: 100
+
+# 
+# io-streaming rate limiting
+# When setting this value to be non-zero scylla throttles disk throughput for
+# stream (network) activities such as backup, repair, tablet migration and more.
+# This limit is useful for user queries so the network interface does 
+# not get saturated by streaming activities.
+# The recommended value is 75% of network bandwidth
+# E.g for i4i.8xlarge (https://github.com/scylladb/scylla-machine-image/tree/next/common/aws_net_params.json):
+# network: 18.75 GiB/s --> 18750 Mib/s --> 1875 MB/s (from network bits to network bytes: divide by 10, not 8)
+# Converted to disk bytes: 1875 * 1000 / 1024 = 1831 MB/s (disk wise)
+# 75% of disk bytes is: 0.75 * 1831 = 1373 megabytes/s
+# stream_io_throughput_mb_per_sec: 1373
+# 
+

cql3/statements/alter_keyspace_statement.cc

@@ -245,12 +245,18 @@ cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_proce
             qp.db().real_database().validate_keyspace_update(*ks_md_update);
 
             service::topology_mutation_builder builder(ts);
+            service::topology_request_tracking_mutation_builder rtbuilder{global_request_id, qp.proxy().features().topology_requests_type_column};
+            rtbuilder.set("done", false)
+                     .set("start_time", db_clock::now());
             if (!qp.proxy().features().topology_global_request_queue) {
                 builder.set_global_topology_request(service::global_topology_request::keyspace_rf_change);
                 builder.set_global_topology_request_id(global_request_id);
                 builder.set_new_keyspace_rf_change_data(_name, ks_options);
             } else {
                 builder.queue_global_topology_request_id(global_request_id);
+                rtbuilder.set("request_type", service::global_topology_request::keyspace_rf_change)
+                         .set_new_keyspace_rf_change_data(_name, ks_options);
+
             };
             service::topology_change change{{builder.build()}};
 
@@ -259,13 +265,6 @@ cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_proce
                 return cm.to_mutation(topo_schema);
             });
 
-            service::topology_request_tracking_mutation_builder rtbuilder{global_request_id, qp.proxy().features().topology_requests_type_column};
-            rtbuilder.set("done", false)
-                     .set("start_time", db_clock::now())
-                     .set("request_type", service::global_topology_request::keyspace_rf_change);
-            if (qp.proxy().features().topology_global_request_queue) {
-                rtbuilder.set_new_keyspace_rf_change_data(_name, ks_options);
-            }
             service::topology_change req_change{{rtbuilder.build()}};
 
             auto topo_req_schema = qp.db().find_schema(db::system_keyspace::NAME, db::system_keyspace::TOPOLOGY_REQUESTS);

cql3/statements/alter_table_statement.cc

@@ -8,6 +8,7 @@
  * SPDX-License-Identifier: (LicenseRef-ScyllaDB-Source-Available-1.0 and Apache-2.0)
  */
 
+#include "cdc/log.hh"
 #include "utils/assert.hh"
 #include <seastar/core/coroutine.hh>
 #include "cql3/query_options.hh"
@@ -27,6 +28,7 @@
 #include "db/view/view.hh"
 #include "cql3/query_processor.hh"
 #include "cdc/cdc_extension.hh"
+#include "cdc/cdc_partitioner.hh"
 
 namespace cql3 {
 
@@ -290,6 +292,53 @@ std::pair<schema_ptr, std::vector<view_ptr>> alter_table_statement::prepare_sche
         throw exceptions::invalid_request_exception("Cannot use ALTER TABLE on Materialized View");
     }
 
+    const bool is_cdc_log_table = cdc::is_log_for_some_table(db.real_database(), s->ks_name(), s->cf_name());
+    // Only a CDC log table will have this partitioner name. User tables should
+    // not be able to set this. Note that we perform a similar check when trying to
+    // re-enable CDC for a table, when the log table has been replaced by a user table.
+    // For better visualization of the above, consider this
+    //
+    // cqlsh> CREATE TABLE ks.t (p int PRIMARY KEY, v int) WITH cdc = {'enabled': true};
+    // cqlsh> INSERT INTO ks.t (p, v) VALUES (1, 2);
+    // cqlsh> ALTER TABLE ks.t WITH cdc = {'enabled': false};
+    // cqlsh> DESC TABLE ks.t_scylla_cdc_log WITH INTERNALS; # Save this output!
+    // cqlsh> DROP TABLE ks.t_scylla_cdc_log;
+    // cqlsh> [Recreate the log table using the received statement]
+    // cqlsh> ALTER TABLE ks.t WITH cdc = {'enabled': true};
+    //
+    // InvalidRequest: Error from server: code=2200 [Invalid query] message="Cannot create CDC log
+    //                 table for table ks.t because a table of name ks.t_scylla_cdc_log already exists"
+    //
+    // See commit adda43edc75b901b2329bca8f3eb74596698d05f for more information on THAT case.
+    // We reuse the same technique here.
+    const bool was_cdc_log_table = s->get_partitioner().name() == cdc::cdc_partitioner::classname;
+
+    if (_column_changes.size() != 0 && is_cdc_log_table) {
+        throw exceptions::invalid_request_exception(
+                "You cannot modify the set of columns of a CDC log table directly. "
+                "Modify the base table instead.");
+    }
+    if (_column_changes.size() != 0 && was_cdc_log_table) {
+        throw exceptions::invalid_request_exception(
+                "You cannot modify the set of columns of a CDC log table directly. "
+                "Although the base table has deactivated CDC, this table will continue being "
+                "a CDC log table until it is dropped. If you want to modify the columns in it, "
+                "you can only do that by reenabling CDC on the base table, which will reattach "
+                "this log table. Then you will be able to modify the columns in the base table, "
+                "and that will have effect on the log table too. Modifying the columns of a CDC "
+                "log table directly is never allowed.");
+    }
+
+    if (_renames.size() != 0 && is_cdc_log_table) {
+        throw exceptions::invalid_request_exception("Cannot rename a column of a CDC log table.");
+    }
+    if (_renames.size() != 0 && was_cdc_log_table) {
+        throw exceptions::invalid_request_exception(
+                "You cannot rename a column of a CDC log table. Although the base table "
+                "has deactivated CDC, this table will continue being a CDC log table until it "
+                "is dropped.");
+    }
+
     auto cfm = schema_builder(s);
 
     if (_properties->get_id()) {

db/batchlog_manager.cc

@@ -36,7 +36,7 @@
 
 static logging::logger blogger("batchlog_manager");
 
-const uint32_t db::batchlog_manager::replay_interval;
+const std::chrono::seconds db::batchlog_manager::replay_interval;
 const uint32_t db::batchlog_manager::page_size;
 
 db::batchlog_manager::batchlog_manager(cql3::query_processor& qp, db::system_keyspace& sys_ks, batchlog_manager_config config)
@@ -116,7 +116,8 @@ future<> db::batchlog_manager::batchlog_replay_loop() {
         } catch (...) {
             blogger.error("Exception in batch replay: {}", std::current_exception());
         }
-        delay = std::chrono::milliseconds(replay_interval);
+        delay = utils::get_local_injector().is_enabled("short_batchlog_manager_replay_interval") ?
+                std::chrono::seconds(1) : replay_interval;
     }
 }
 
@@ -132,6 +133,8 @@ future<> db::batchlog_manager::drain() {
         _sem.broken();
     }
 
+    co_await _qp.proxy().abort_batch_writes();
+
     co_await std::move(_loop_done);
     blogger.info("Drained");
 }
@@ -173,6 +176,11 @@ future<> db::batchlog_manager::replay_all_failed_batches(post_replay_cleanup cle
             return make_ready_future<stop_iteration>(stop_iteration::no);
         }
 
+        if (utils::get_local_injector().is_enabled("skip_batch_replay")) {
+            blogger.debug("Skipping batch replay due to skip_batch_replay injection");
+            return make_ready_future<stop_iteration>(stop_iteration::no);
+        }
+
         // check version of serialization format
         if (!row.has("version")) {
             blogger.warn("Skipping logged batch because of unknown version");
@@ -242,7 +250,8 @@ future<> db::batchlog_manager::replay_all_failed_batches(post_replay_cleanup cle
                 // send to partially or wholly fail in actually sending stuff. Since we don't
                 // have hints (yet), send with CL=ALL, and hope we can re-do this soon.
                 // See below, we use retry on write failure.
-                return _qp.proxy().mutate(mutations, db::consistency_level::ALL, db::no_timeout, nullptr, empty_service_permit(), db::allow_per_partition_rate_limit::no);
+                auto timeout = db::timeout_clock::now() + write_timeout;
+                return _qp.proxy().send_batchlog_replay_to_all_replicas(std::move(mutations), timeout);
             });
         }).then_wrapped([this, id](future<> batch_result) {
             try {

db/batchlog_manager.hh

@@ -43,8 +43,9 @@ public:
     using post_replay_cleanup = bool_class<class post_replay_cleanup_tag>;
 
 private:
-    static constexpr uint32_t replay_interval = 60 * 1000; // milliseconds
+    static constexpr std::chrono::seconds replay_interval = std::chrono::seconds(60);
     static constexpr uint32_t page_size = 128; // same as HHOM, for now, w/out using any heuristics. TODO: set based on avg batch size.
+    static constexpr std::chrono::seconds write_timeout = std::chrono::seconds(300);
 
     using clock_type = lowres_clock;
 

db/config.cc

@@ -86,6 +86,12 @@ object_storage_endpoints_to_json(const std::vector<db::object_storage_endpoint_p
     return value_to_json(m);
 }
 
+static
+json::json_return_type
+uuid_to_json(const db::config::UUID& uuid) {
+    return value_to_json(format("{}", uuid));
+}
+
 // Convert a value that can be printed with fmt::format, or a vector of
 // such values, to JSON. An example is enum_option<T>, because enum_option<T>
 // has a specialization for fmt::formatter.
@@ -294,6 +300,12 @@ const config_type& config_type_for<std::vector<db::object_storage_endpoint_param
     return ct;
 }
 
+template <>
+const config_type& config_type_for<db::config::UUID>() {
+    static config_type ct("UUID", uuid_to_json);
+    return ct;
+}
+
 }
 
 namespace YAML {
@@ -491,6 +503,22 @@ struct convert<db::object_storage_endpoint_param> {
     }
 };
 
+template<>
+struct convert<utils::UUID> {
+    static bool decode(const Node& node, utils::UUID& uuid) {
+        std::string uuid_string;
+        if (!convert<std::string>::decode(node, uuid_string)) {
+            return false;
+        }
+        try {
+            std::istringstream(uuid_string) >> uuid;
+        } catch (boost::program_options::invalid_option_value&) {
+            return false;
+        }
+        return true;
+    }
+};
+
 }
 
 #if defined(DEBUG)
@@ -819,7 +847,7 @@ db::config::config(std::shared_ptr<db::extensions> exts)
     , inter_dc_stream_throughput_outbound_megabits_per_sec(this, "inter_dc_stream_throughput_outbound_megabits_per_sec", value_status::Unused, 0,
         "Throttles all streaming file transfer between the data centers. This setting allows throttles streaming throughput betweens data centers in addition to throttling all network stream traffic as configured with stream_throughput_outbound_megabits_per_sec.")
     , stream_io_throughput_mb_per_sec(this, "stream_io_throughput_mb_per_sec", liveness::LiveUpdate, value_status::Used, 0,
-        "Throttles streaming I/O to the specified total throughput (in MiBs/s) across the entire system. Streaming I/O includes the one performed by repair and both RBNO and legacy topology operations such as adding or removing a node. Setting the value to 0 disables stream throttling.")
+        "Throttles streaming I/O to the specified total throughput (in MiBs/s) across the entire system. Streaming I/O includes the one performed by repair and both RBNO and legacy topology operations such as adding or removing a node. Setting the value to 0 disables stream throttling. It is recommended to set the value for this parameter to be 75% of network bandwidth")
     , stream_plan_ranges_fraction(this, "stream_plan_ranges_fraction", liveness::LiveUpdate, value_status::Used, 0.1,
         "Specify the fraction of ranges to stream in a single stream plan. Value is between 0 and 1.")
     , enable_file_stream(this, "enable_file_stream", liveness::LiveUpdate, value_status::Used, true, "Set true to use file based stream for tablet instead of mutation based stream")
@@ -1230,7 +1258,7 @@ db::config::config(std::shared_ptr<db::extensions> exts)
     , sstable_summary_ratio(this, "sstable_summary_ratio", value_status::Used, 0.0005, "Enforces that 1 byte of summary is written for every N (2000 by default)"
         "bytes written to data file. Value must be between 0 and 1.")
     , components_memory_reclaim_threshold(this, "components_memory_reclaim_threshold", liveness::LiveUpdate, value_status::Used, .2, "Ratio of available memory for all in-memory components of SSTables in a shard beyond which the memory will be reclaimed from components until it falls back under the threshold. Currently, this limit is only enforced for bloom filters.")
-    , large_memory_allocation_warning_threshold(this, "large_memory_allocation_warning_threshold", value_status::Used, (size_t(128) << 10) + 1, "Warn about memory allocations above this size; set to zero to disable.")
+    , large_memory_allocation_warning_threshold(this, "large_memory_allocation_warning_threshold", value_status::Used, size_t(1) << 20, "Warn about memory allocations above this size; set to zero to disable.")
     , enable_deprecated_partitioners(this, "enable_deprecated_partitioners", value_status::Used, false, "Enable the byteordered and random partitioners. These partitioners are deprecated and will be removed in a future version.")
     , enable_keyspace_column_family_metrics(this, "enable_keyspace_column_family_metrics", value_status::Used, false, "Enable per keyspace and per column family metrics reporting.")
     , enable_node_aggregated_table_metrics(this, "enable_node_aggregated_table_metrics", value_status::Used, true, "Enable aggregated per node, per keyspace and per table metrics reporting, applicable if enable_keyspace_column_family_metrics is false.")
@@ -1399,7 +1427,7 @@ db::config::config(std::shared_ptr<db::extensions> exts)
         "The maximum fraction of cache memory permitted for use by index cache. Clamped to the [0.0; 1.0] range. Must be small enough to not deprive the row cache of memory, but should be big enough to fit a large fraction of the index. The default value 0.2 means that at least 80\% of cache memory is reserved for the row cache, while at most 20\% is usable by the index cache.")
     , consistent_cluster_management(this, "consistent_cluster_management", value_status::Deprecated, true, "Use RAFT for cluster management and DDL.")
     , force_gossip_topology_changes(this, "force_gossip_topology_changes", value_status::Used, false, "Force gossip-based topology operations in a fresh cluster. Only the first node in the cluster must use it. The rest will fall back to gossip-based operations anyway. This option should be used only for testing.  Note: gossip topology changes are incompatible with tablets.")
-    , recovery_leader(this, "recovery_leader", liveness::LiveUpdate, value_status::Used, "", "Host ID of the node restarted first while performing the Manual Raft-based Recovery Procedure. Warning: this option disables some guardrails for the needs of the Manual Raft-based Recovery Procedure. Make sure you unset it at the end of the procedure.")
+    , recovery_leader(this, "recovery_leader", liveness::LiveUpdate, value_status::Used, utils::null_uuid(), "Host ID of the node restarted first while performing the Manual Raft-based Recovery Procedure. Warning: this option disables some guardrails for the needs of the Manual Raft-based Recovery Procedure. Make sure you unset it at the end of the procedure.")
     , wasm_cache_memory_fraction(this, "wasm_cache_memory_fraction", value_status::Used, 0.01, "Maximum total size of all WASM instances stored in the cache as fraction of total shard memory.")
     , wasm_cache_timeout_in_ms(this, "wasm_cache_timeout_in_ms", value_status::Used, 5000, "Time after which an instance is evicted from the cache.")
     , wasm_cache_instance_size_limit(this, "wasm_cache_instance_size_limit", value_status::Used, 1024*1024, "Instances with size above this limit will not be stored in the cache.")

db/config.hh

@@ -207,6 +207,7 @@ public:
     using seed_provider_type = db::seed_provider_type;
     using hinted_handoff_enabled_type = db::hints::host_filter;
     using error_injection_at_startup = db::error_injection_at_startup;
+    using UUID = utils::UUID;
 
     /*
      * All values and documentation taken from
@@ -521,7 +522,7 @@ public:
 
     named_value<bool> consistent_cluster_management;
     named_value<bool> force_gossip_topology_changes;
-    named_value<sstring> recovery_leader;
+    named_value<UUID> recovery_leader;
 
     named_value<double> wasm_cache_memory_fraction;
     named_value<uint32_t> wasm_cache_timeout_in_ms;

db/system_keyspace.cc

@@ -1694,6 +1694,12 @@ future<> system_keyspace::peers_table_read_fixup() {
             continue;
         }
         const auto host_id = row.get_as<utils::UUID>("host_id");
+        if (!host_id) {
+            slogger.error("Peer {} has null host_id in system.{}, the record is broken, removing it",
+                peer, system_keyspace::PEERS);
+            co_await remove_endpoint(gms::inet_address{peer});
+            continue;
+        }
         const auto ts = row.get_as<int64_t>("ts");
         const auto it = map.find(host_id);
         if (it == map.end()) {

dist/common/scripts/scylla_sysconfig_setup

@@ -86,9 +86,9 @@ if __name__ == '__main__':
     ethpciid = ''
     if network_mode == 'dpdk':
         dpdk_status = out('/opt/scylladb/scripts/dpdk-devbind.py --status')
-        match = re.search('if={} drv=(\S+)'.format(ifname), dpdk_status, flags=re.MULTILINE)
+        match = re.search(r'if={} drv=(\S+)'.format(ifname), dpdk_status, flags=re.MULTILINE)
         ethdrv = match.group(1)
-        match = re.search('^(\\S+:\\S+:\\S+\.\\S+) [^\n]+ if={} '.format(ifname), dpdk_status, flags=re.MULTILINE)
+        match = re.search(r'^(\S+:\S+:\S+\.\S+) [^\n]+ if={} '.format(ifname), dpdk_status, flags=re.MULTILINE)
         ethpciid = match.group(1)
 
     if args.mode:

dist/debian/control.template

@@ -18,7 +18,7 @@ Breaks: scylla-enterprise-conf (<< 2025.1.0~)
 
 Package: %{product}-server
 Architecture: any
-Depends: ${misc:Depends}, %{product}-conf (= ${binary:Version}), %{product}-python3 (= ${binary:Version})
+Depends: ${misc:Depends}, %{product}-conf (= ${binary:Version}), %{product}-python3 (= ${binary:Version}), procps
 Replaces: %{product}-tools (<<5.5), scylla-enterprise-tools (<< 2024.2.0~), scylla-enterprise-server (<< 2025.1.0~)
 Breaks: %{product}-tools (<<5.5), scylla-enterprise-tools (<< 2024.2.0~), scylla-enterprise-server (<< 2025.1.0~)
 Description: Scylla database server binaries

dist/docker/redhat/build_docker.sh

@@ -88,7 +88,7 @@ bcp LICENSE-ScyllaDB-Source-Available.md /licenses/
 
 run microdnf clean all
 run microdnf --setopt=tsflags=nodocs -y update
-run microdnf --setopt=tsflags=nodocs -y install hostname python3 python3-pip kmod
+run microdnf --setopt=tsflags=nodocs -y install hostname kmod procps-ng python3 python3-pip
 run microdnf clean all
 run pip3 install --no-cache-dir --prefix /usr supervisor
 run bash -ec "echo LANG=C.UTF-8 > /etc/locale.conf"

dist/redhat/scylla.spec

@@ -76,6 +76,7 @@ Group:          Applications/Databases
 Summary:        The Scylla database server
 Requires:       %{product}-conf = %{version}-%{release}
 Requires:       %{product}-python3 = %{version}-%{release}
+Requires:       procps-ng
 AutoReqProv:    no
 Provides:       %{product}-tools:%{_bindir}/nodetool
 Provides:       %{product}-tools:%{_sysconfigdir}/bash_completion.d/nodetool-completion

docs/_static/data/os-support.json

@@ -2,10 +2,19 @@
     "Linux Distributions": {
       "Ubuntu": ["22.04", "24.04"],
       "Debian": ["11"],
-      "Rocky / CentOS / RHEL": ["8", "9"],
+      "Rocky / CentOS / RHEL": ["8", "9", "10"],
       "Amazon Linux": ["2023"]
     },
     "ScyllaDB Versions": [
+      {
+        "version": "ScyllaDB 2025.3",
+        "supported_OS": {
+          "Ubuntu": ["22.04", "24.04"],
+          "Debian": ["11"],
+          "Rocky / CentOS / RHEL": ["8", "9", "10"],
+          "Amazon Linux": ["2023"]
+        }
+      },
       {
         "version": "ScyllaDB 2025.2",
         "supported_OS": {

docs/_utils/redirects.yaml

@@ -172,4 +172,7 @@
 /stable/upgrade/upgrade-opensource/upgrade-guide-from-4.5-to-4.6/metric-update-4.5-to-4.6.html: /stable/upgrade/index.html
 
 # Divide API reference to smaller files
-# /stable/reference/api-reference.html: /stable/reference/api/index.html
+# /stable/reference/api-reference.html: /stable/reference/api/index.html
+
+# Fixed typo in the file name
+/stable/operating-scylla/nodetool-commands/enbleautocompaction.html: /stable/operating-scylla/nodetool-commands/enableautocompaction.html

docs/cql/types.rst

@@ -481,7 +481,8 @@ Creating a new user-defined type is done using a ``CREATE TYPE`` statement defin
    field_definition: `identifier` `cql_type`
 
 A UDT has a name (``udt_name``), which is used to declare columns of that type and is a set of named and typed fields. The ``udt_name`` can be any
-type, including collections or other UDTs. UDTs and collections inside collections must always be frozen (no matter which version of ScyllaDB you are using). 
+type, including collections or other UDTs.
+Similar to collections, a UDT can be frozen or non-frozen. A frozen UDT is immutable and can only be updated as a whole. Nested UDTs or UDTs used in keys must always be frozen.
 
 For example::
 
@@ -506,26 +507,15 @@ For example::
 
   CREATE TABLE superheroes (
        name frozen<full_name> PRIMARY KEY,
-       home frozen<address>
+       home address
   );
 
 .. note::
 
    - Attempting to create an already existing type will result in an error unless the ``IF NOT EXISTS`` option is used. If it is used, the statement will be a no-op if the type already exists.
    - A type is intrinsically bound to the keyspace in which it is created and can only be used in that keyspace. At creation, if the type name is prefixed by a keyspace name, it is created in that keyspace. Otherwise, it is created in the current keyspace.
-   - As of ScyllaDB Open Source 3.2, UDTs not inside collections do not have to be frozen, but in all versions prior to ScyllaDB Open Source 3.2, and in all ScyllaDB Enterprise versions, UDTs **must** be frozen. 
 
 
-A non-frozen UDT example with ScyllaDB Open Source 3.2 and higher::
-
-   CREATE TYPE ut (a int, b int);
-   CREATE TABLE cf (a int primary key, b ut);
-
-Same UDT in versions prior::
-
-   CREATE TYPE ut (a int, b int);
-   CREATE TABLE cf (a int primary key, b frozen<ut>);
-
 UDT literals
 ~~~~~~~~~~~~
 

docs/dev/testing.md

@@ -10,7 +10,7 @@ This is a manual for `test.py`.
 
 ## Installation
 
-To run `test.py`, Python 3.7 or higher is required.
+To run `test.py`, Python 3.11 or higher is required.
 `./install-dependencies.sh` should install all the required Python
 modules. If `install-dependencies.sh` does not support your distribution,
 please manually install all Python modules it lists with `pip`.
@@ -106,6 +106,19 @@ shed more light on this.
 Build artefacts, such as test output and harness output is stored
 in `./testlog`. Scylla data files are stored in `/tmp`.
 
+There are several test directories that are excluded from orchestration by `test.py`:
+
+- test/boost
+- test/raft
+- test/ldap
+- test/unit
+
+This means that `test.py` will not run tests directly, but will delegate all work to `pytest`.
+That's why all these directories do not have `suite.yaml` files.
+Additionally, these directories do not follow abstract naming suite/testname
+convention, and instead use the `pytest` naming convention, i.e. to run a test you need to provide the path to the file
+and optionally the test name, e.g. `test/boost/aggregate_fcts_test.cc::test_aggregate_avg`.
+
 ## How it works
 
 On start, `test.py` invokes `ninja` to find out configured build modes. Then
@@ -176,11 +189,11 @@ Scylla (possibly started in debugger) using `cqlsh`.
 
 The same unit test can be run in different seastar configurations, i.e. with
 different command line arguments. The custom arguments can be set in
-`custom_args` key of the `suite.yaml` file.
+`custom_args` key of the `test_config.yaml` file.
 
 Tests from boost suite are divided into test-cases. These are top-level
 functions wrapped by `BOOST_AUTO_TEST_CASE`, `SEASTAR_TEST_CASE` or alike.
-Boost tests support `suitename/testname::casename` selection described above.
+Boost tests support `path/to/file_name.cc::casename` selection described above.
 
 ### Debugging unit tests
 
@@ -328,6 +341,19 @@ as it was at the beginning of the test, is considered "dirty".
 Such clusters are not returned to the pool, but destroyed, and
 the pool is replenished with a new cluster instead.
 
+## Test metrics
+
+The parameter `--gather-metrics` is used to gather CPU/RAM usage during tests from the cgroup and system overall CPU/RAM
+usage.
+For that, SQLite database is used to store the metrics in `testlog/sqlite.db`.
+The database is created in the `testlog` directory and contains the following tables:
+
+- `tests` - contains the list of tests that were executed with information about the test name, directory, architecture,
+  and mode
+- `test_metrics` - contains the metrics for each test, such as memory peak usage, CPU usage, and duration
+- `system_resource_metrics` - contains system CPU and memory utilization in percents during the whole run
+- `cgroup_memory_metrics` - contains cgroup memory usage during the test run
+
 ## Automation, CI, and Jenkins
 
 If any of the tests fails, `test.py` returns a non-zero exit status.

docs/getting-started/cloud-instance-recommendations.rst

@@ -113,8 +113,8 @@ Pick a zone where Haswell CPUs are found. Local SSD performance offers, accordin
 Image with NVMe disk interface is recommended.
 (`More info <https://cloud.google.com/compute/docs/disks/local-ssd>`_)
 
-Recommended instances types are `z3-highmem-highlssd <https://cloud.google.com/compute/docs/storage-optimized-machines#z3_machine_types>`_,
-`n1-highmem <https://cloud.google.com/compute/docs/general-purpose-machines#n1_machines>`_, and `n2-highmem <https://cloud.google.com/compute/docs/general-purpose-machines#n2_machines>`_
+Recommended instances types are `z3-highmem-highlssd and z3-highmem-standardlssd <https://cloud.google.com/compute/docs/storage-optimized-machines#z3_machine_types>`_,
+`n1-highmem <https://cloud.google.com/compute/docs/general-purpose-machines#n1_machines>`_, and `n2-highmem <https://cloud.google.com/compute/docs/general-purpose-machines#n2_machines>`_.
 
 
 .. list-table::
@@ -145,6 +145,39 @@ Recommended instances types are `z3-highmem-highlssd <https://cloud.google.com/c
      - 44
      - 352
      - 18,000
+   * - z3-highmem-88-highlssd
+     - 88
+     - 704
+     - 36,000	  
+
+.. list-table::
+   :widths: 30 20 20 30
+   :header-rows: 1
+
+   * - Model
+     - vCPU
+     - Mem (GB)
+     - Storage (GB)
+   * - z3-highmem-14-standardlssd
+     - 14
+     - 112
+     - 3,000
+   * - z3-highmem-22-standardlssd
+     - 22
+     - 176
+     - 6,000
+   * - z3-highmem-44-standardlssd 
+     - 44
+     - 352
+     - 9,000
+   * - z3-highmem-88-standardlssd
+     - 88
+     - 704
+     - 18,000
+   * - z3-highmem-176-standardlssd
+     - 176
+     - 1,406
+     - 36,000 
 
 .. list-table::
    :widths: 30 20 20 30

docs/kb/consistency.rst

@@ -83,7 +83,7 @@ Additional References
 
 * `Jepsen and ScyllaDB: Putting Consistency to the Test blog post <https://www.scylladb.com/2020/12/23/jepsen-and-scylla-putting-consistency-to-the-test/>`_ 
 * `Nauto: Achieving Consistency in an Eventually Consistent Environment blog post <https://www.scylladb.com/2020/02/20/nauto-achieving-consistency-in-an-eventually-consistent-environment/>`_ 
-* `Consistency Levels documentation <https://docs.scylladb.com/stable/cql/consistency.html>`_ 
+* `Consistency Levels documentation <https://docs.scylladb.com/manual/stable/cql/consistency.html>`_ 
 * `High Availability lesson on ScyllaDB University <https://university.scylladb.com/courses/scylla-essentials-overview/lessons/high-availability/>`_ 
 * `Lightweight Transactions lesson on ScyllaDB University <https://university.scylladb.com/courses/data-modeling/lessons/lightweight-transactions/>`_ 
 * `Getting the Most out of Lightweight Transactions in ScyllaDB blog post <https://www.scylladb.com/2020/07/15/getting-the-most-out-of-lightweight-transactions-in-scylla/>`_ 

docs/operating-scylla/nodetool-commands/restore.rst

@@ -26,6 +26,7 @@ Syntax
                --table <table>
                [--nowait]
                [--scope <scope>]
+               [--sstables-file-list <file>]
                <sstables>...
 
 Example
@@ -51,6 +52,7 @@ Options
 * ``--table`` - Name of the table to load SSTables into
 * ``--nowait`` - Don't wait on the restore process
 * ``--scope <scope>`` - Use specified load-and-stream scope
+* ``--sstables-file-list <file>`` - restore the sstables listed in the given <file>. the list should be new-line seperated.
 * ``<sstables>`` - Remainder of keys of the TOC (Table of Contents) components of SSTables to restore, relative to the specified prefix
 
 The `scope` parameter describes the subset of cluster nodes where you want to load data:
@@ -60,6 +62,8 @@ The `scope` parameter describes the subset of cluster nodes where you want to lo
 * `dc` - In the datacenter (DC) where the local node lives.
 * `all` (default) - Everywhere across the cluster.
 
+`--sstables-file-list <file>` and `<sstable>` can be combined together, `nodetool restore` will attempt to restore the combined list. duplicates are _not_ removed
+
 To fully restore a cluster, you should combine the ``scope`` parameter with the correct list of
 SStables to restore to each node.
 On one extreme, one node is given all SStables with the scope ``all``; on the other extreme, all

docs/operating-scylla/nodetool.rst

@@ -14,9 +14,9 @@ Nodetool
    nodetool-commands/cleanup
    nodetool-commands/clearsnapshot
    nodetool-commands/cluster/index
+   nodetool-commands/compact
    nodetool-commands/compactionhistory
    nodetool-commands/compactionstats
-   nodetool-commands/compact
    nodetool-commands/decommission
    nodetool-commands/describecluster
    nodetool-commands/describering
@@ -25,13 +25,15 @@ Nodetool
    nodetool-commands/disablebinary
    nodetool-commands/disablegossip
    nodetool-commands/drain
-   nodetool-commands/enbleautocompaction   
+   nodetool-commands/enableautocompaction
    nodetool-commands/enablebackup
    nodetool-commands/enablebinary
    nodetool-commands/enablegossip
    nodetool-commands/flush
+   nodetool-commands/getcompactionthroughput
    nodetool-commands/getendpoints
    nodetool-commands/getsstables
+   nodetool-commands/getstreamthroughput
    nodetool-commands/gettraceprobability
    nodetool-commands/gossipinfo
    nodetool-commands/help
@@ -46,25 +48,23 @@ Nodetool
    nodetool-commands/restore
    nodetool-commands/ring
    nodetool-commands/scrub
-   nodetool-commands/settraceprobability
+   nodetool-commands/setcompactionthroughput
    nodetool-commands/setlogginglevel
+   nodetool-commands/setstreamthroughput
+   nodetool-commands/settraceprobability
    nodetool-commands/snapshot
    nodetool-commands/sstableinfo
+   nodetool-commands/status
    nodetool-commands/statusbackup
    nodetool-commands/statusbinary
    nodetool-commands/statusgossip
-   nodetool-commands/status
    Nodetool stop compaction <nodetool-commands/stop>
    nodetool-commands/tablestats
    nodetool-commands/tasks/index
    nodetool-commands/toppartitions
    nodetool-commands/upgradesstables
-   nodetool-commands/viewbuildstatus
    nodetool-commands/version
-   nodetool-commands/getcompactionthroughput
-   nodetool-commands/setcompactionthroughput
-   nodetool-commands/getstreamthroughput
-   nodetool-commands/setstreamthroughput
+   nodetool-commands/viewbuildstatus
 
 The ``nodetool`` utility provides a simple command-line interface to the following exposed operations and attributes.
 
@@ -87,9 +87,9 @@ Operations that are not listed below are currently not available.
 * :doc:`cleanup </operating-scylla/nodetool-commands/cleanup/>` - Triggers the immediate cleanup of keys no longer belonging to a node.
 * :doc:`clearsnapshot </operating-scylla/nodetool-commands/clearsnapshot/>` - This command removes snapshots.
 * :doc:`cluster <nodetool-commands/cluster/index>` - Run a cluster operation.
+* :doc:`compact </operating-scylla/nodetool-commands/compact/>`- Force a (major) compaction on one or more column families.
 * :doc:`compactionhistory </operating-scylla/nodetool-commands/compactionhistory/>` - Provides the history of compactions.
 * :doc:`compactionstats </operating-scylla/nodetool-commands/compactionstats/>`- Print statistics on compactions.
-* :doc:`compact </operating-scylla/nodetool-commands/compact/>`- Force a (major) compaction on one or more column families.
 * :doc:`decommission </operating-scylla/nodetool-commands/decommission/>` - Decommission the node.
 * :doc:`describecluster </operating-scylla/nodetool-commands/describecluster/>` - Print the name, snitch, partitioner and schema version of a cluster.
 * :doc:`describering </operating-scylla/nodetool-commands/describering/>` - :code:`<keyspace>`- Shows the partition ranges of a given keyspace.
@@ -98,14 +98,16 @@ Operations that are not listed below are currently not available.
 * :doc:`disablebinary </operating-scylla/nodetool-commands/disablebinary/>` - Disable native transport (binary protocol).
 * :doc:`disablegossip </operating-scylla/nodetool-commands/disablegossip/>` - Disable gossip (effectively marking the node down).
 * :doc:`drain </operating-scylla/nodetool-commands/drain/>` - Drain the node (stop accepting writes and flush all column families).
-* :doc:`enableautocompaction </operating-scylla/nodetool-commands/enbleautocompaction/>` - Enable automatic compaction of a keyspace or table.
+* :doc:`enableautocompaction </operating-scylla/nodetool-commands/enableautocompaction/>` - Enable automatic compaction of a keyspace or table.
 * :doc:`enablebackup </operating-scylla/nodetool-commands/enablebackup/>` - Enable incremental backup.
 * :doc:`enablebinary </operating-scylla/nodetool-commands/enablebinary/>` - Re-enable native transport (binary protocol).
 * :doc:`enablegossip </operating-scylla/nodetool-commands/enablegossip/>` - Re-enable gossip.
 * :doc:`flush </operating-scylla/nodetool-commands/flush/>` - Flush one or more column families.
+* :doc:`getcompactionthroughput </operating-scylla/nodetool-commands/getcompactionthroughput>` - Print the throughput cap for compaction in the system
 * :doc:`getendpoints <nodetool-commands/getendpoints/>` :code:`<keyspace>` :code:`<table>` :code:`<key>`- Print the end points that owns the key.
 * **getlogginglevels** - Get the runtime logging levels.
 * :doc:`getsstables </operating-scylla/nodetool-commands/getsstables>` - Print the sstable filenames that own the key.
+* :doc:`getstreamthroughput </operating-scylla/nodetool-commands/getstreamthroughput>` - Print the throughput cap for SSTables streaming in the system
 * :doc:`gettraceprobability </operating-scylla/nodetool-commands/gettraceprobability>` - Displays the current trace probability value. 0 is disabled 1 is enabled.
 * :doc:`gossipinfo </operating-scylla/nodetool-commands/gossipinfo/>` - Shows the gossip information for the cluster.
 * :doc:`help </operating-scylla/nodetool-commands/help/>` - Display list of available nodetool commands.
@@ -118,28 +120,26 @@ Operations that are not listed below are currently not available.
 * :doc:`refresh </operating-scylla/nodetool-commands/refresh/>`- Load newly placed SSTables to the system without restart
 * :doc:`removenode </operating-scylla/nodetool-commands/removenode/>`- Remove node with the provided ID
 * :doc:`repair <nodetool-commands/repair/>`  :code:`<keyspace>` :code:`<table>` - Repair one or more vnode tables.
-* :doc:`restore </operating-scylla/nodetool-commands/restore/>` - Load SSTables from a designated bucket in object store into a specified keyspace or table
 * :doc:`resetlocalschema </operating-scylla/nodetool-commands/resetlocalschema/>` - Reset the node's local schema.
+* :doc:`restore </operating-scylla/nodetool-commands/restore/>` - Load SSTables from a designated bucket in object store into a specified keyspace or table
 * :doc:`ring <nodetool-commands/ring/>` - The nodetool ring command display the token ring information.
 * :doc:`scrub </operating-scylla/nodetool-commands/scrub>` :code:`[-m mode] [--no-snapshot] <keyspace> [<table>...]` - Scrub the SSTable files in the specified keyspace or table(s)
+* :doc:`setcompactionthroughput </operating-scylla/nodetool-commands/setcompactionthroughput>` - Set the throughput cap for compaction in the system
 * :doc:`setlogginglevel</operating-scylla/nodetool-commands/setlogginglevel>` - sets the logging level threshold for ScyllaDB classes
+* :doc:`setstreamthroughput </operating-scylla/nodetool-commands/setstreamthroughput>` - Set the throughput cap for SSTables streaming in the system
 * :doc:`settraceprobability </operating-scylla/nodetool-commands/settraceprobability/>` ``<value>`` - Sets the probability for tracing a request. race probability value
 * :doc:`snapshot </operating-scylla/nodetool-commands/snapshot>` :code:`[-t tag] [-cf column_family] <keyspace>`  - Take a snapshot of specified keyspaces or a snapshot of the specified table.
 * :doc:`sstableinfo </operating-scylla/nodetool-commands/sstableinfo>` - Get information about sstables per keyspace/table.
+* :doc:`status </operating-scylla/nodetool-commands/status/>` - Print cluster information.
 * :doc:`statusbackup </operating-scylla/nodetool-commands/statusbackup/>` - Status of incremental backup.
 * :doc:`statusbinary </operating-scylla/nodetool-commands/statusbinary/>` - Status of native transport (binary protocol).
 * :doc:`statusgossip </operating-scylla/nodetool-commands/statusgossip/>` - Status of gossip.
-* :doc:`status </operating-scylla/nodetool-commands/status/>` - Print cluster information.
 * :doc:`stop </operating-scylla/nodetool-commands/stop/>` - Stop compaction operation.
 * **tablehistograms** see :doc:`cfhistograms <nodetool-commands/cfhistograms/>`
 * :doc:`tablestats </operating-scylla/nodetool-commands/tablestats/>` - Provides in-depth diagnostics regard table. 
 * :doc:`tasks </operating-scylla/nodetool-commands/tasks/index>` - Manage tasks manager tasks.
 * :doc:`toppartitions </operating-scylla/nodetool-commands/toppartitions/>` - Samples cluster writes and reads and reports the most active partitions in a specified table and time frame.
 * :doc:`upgradesstables </operating-scylla/nodetool-commands/upgradesstables>` - Upgrades each table that is not running the latest ScyllaDB version, by rewriting SSTables.
-* :doc:`viewbuildstatus </operating-scylla/nodetool-commands/viewbuildstatus/>` - Shows the progress of a materialized view build.
 * :doc:`version </operating-scylla/nodetool-commands/version>` - Print the DB version.
-* :doc:`getcompactionthroughput </operating-scylla/nodetool-commands/getcompactionthroughput>` - Print the throughput cap for compaction in the system
-* :doc:`setcompactionthroughput </operating-scylla/nodetool-commands/setcompactionthroughput>` - Set the throughput cap for compaction in the system
-* :doc:`getstreamthroughput </operating-scylla/nodetool-commands/getstreamthroughput>` - Print the throughput cap for SSTables streaming in the system
-* :doc:`setstreamthroughput </operating-scylla/nodetool-commands/setstreamthroughput>` - Set the throughput cap for SSTables streaming in the system
+* :doc:`viewbuildstatus </operating-scylla/nodetool-commands/viewbuildstatus/>` - Shows the progress of a materialized view build.
 

docs/troubleshooting/handling-node-failures.rst

@@ -148,16 +148,25 @@ will leave the recovery mode and remove the obsolete internal Raft data.
         cqlsh> TRUNCATE TABLE system.discovery;
         cqlsh> DELETE value FROM system.scylla_local WHERE key = 'raft_group0_id';
 
-#. Add the ``recovery_leader`` property to the ``scylla.yaml`` file and set it to the host ID of the recovery leader on
-   **every live node**. Make sure the change is applied on all nodes by sending the ``SIGHUP`` signal to all ScyllaDB
-   processes.
-
 #. Perform a :doc:`rolling restart </operating-scylla/procedures/config-change/rolling-restart/>` of all live nodes,
-   however, this time **the recovery leader must be restarted first**.
+   but:
+
+   * **restart the recovery leader first**,
+
+   * before restarting each node, add the ``recovery_leader`` property to its ``scylla.yaml`` file and set it to the
+     host ID of the recovery leader,
+
+   * after restarting each node, make sure it participated in Raft recovery; look for one of the following messages
+     in its logs:
+
+    .. code-block:: console
+
+        storage_service - Performing Raft-based recovery procedure with recovery leader <host ID of the recovery leader>/<IP address of the recovery leader>
+        storage_service - Raft-based recovery procedure - found group 0 with ID <ID of the new group 0; different from the one used in other steps>
 
    After completing this step, Raft should be fully functional.
 
-#. Replace all dead nodes from the cluster using the
+#. Replace all dead nodes in the cluster using the
    :doc:`node replacement procedure </operating-scylla/procedures/cluster-management/replace-dead-node/>`.
 
    .. note::

docs/upgrade/upgrade-guides/index.rst

@@ -4,7 +4,7 @@ Upgrade ScyllaDB
 
 .. toctree::
    
-   ScyllaDB 2025.1 to ScyllaDB 2025.2 <upgrade-guide-from-2025.1-to-2025.2/index>
+   ScyllaDB 2025.2 to ScyllaDB 2025.3 <upgrade-guide-from-2025.2-to-2025.3/index>
    ScyllaDB Image <ami-upgrade>
 
 

docs/upgrade/upgrade-guides/upgrade-guide-from-2025.1-to-2025.2/index.rst

@@ -1,13 +0,0 @@
-==========================================================
-Upgrade - ScyllaDB 2025.1 to ScyllaDB 2025.2
-==========================================================
-
-.. toctree::
-   :maxdepth: 2
-   :hidden:
-
-   Upgrade ScyllaDB <upgrade-guide-from-2025.1-to-2025.2>
-   Metrics Update <metric-update-2025.1-to-2025.2>
-
-* :doc:`Upgrade from ScyllaDB 2025.1.x to ScyllaDB 2025.2.y <upgrade-guide-from-2025.1-to-2025.2>`
-* :doc:`Metrics Update Between 2025.1 and 2025.2 <metric-update-2025.1-to-2025.2>`

docs/upgrade/upgrade-guides/upgrade-guide-from-2025.1-to-2025.2/metric-update-2025.1-to-2025.2.rst

@@ -1,61 +0,0 @@
-.. |SRC_VERSION| replace:: 2025.1
-.. |NEW_VERSION| replace:: 2025.2
-
-Metrics Update Between |SRC_VERSION| and |NEW_VERSION|
-================================================================
-
-.. toctree::
-   :maxdepth: 2
-   :hidden:
-
-ScyllaDB |NEW_VERSION| Dashboards are available as part of the latest |mon_root|.
-
-New Metrics
-------------
-
-The following metrics are new in ScyllaDB |NEW_VERSION| compared to |SRC_VERSION|:
-
-
-.. list-table::
-   :widths: 25 150
-   :header-rows: 1
-
-   * - Metric
-     - Description
-   * - scylla_alternator_batch_item_count_histogram
-     - A histogram of the number of items in a batch request.
-   * - scylla_database_total_view_updates_failed_pairing
-     - Total number of view updates for which we failed base/view pairing.
-   * - scylla_group_name_cross_rack_collocations
-     - The number of co-locating migrations that move replica across racks.
-   * - scylla_network_bytes_received
-     - The number of bytes received from network sockets.
-   * - scylla_network_bytes_sent
-     - The number of bytes written to network sockets.
-   * - scylla_reactor_awake_time_ms_total
-     - Total reactor awake time (wall_clock).
-   * - scylla_reactor_cpu_used_time_ms
-     - Total reactor thread CPU time (from CLOCK_THREAD_CPUTIME).
-   * - scylla_reactor_sleep_time_ms_total
-     - Total reactor sleep time (wall clock).
-   * - scylla_sstable_compression_dicts_total_live_memory_bytes
-     - Total amount of memory consumed by SSTable compression dictionaries in RAM.
-   * - scylla_transport_connections_blocked
-     - Holds an incrementing counter with the CQL connections that were blocked
-       before being processed due to threshold configured via
-       uninitialized_connections_semaphore_cpu_concurrency.Blocks are normal
-       when we have multiple connections initialized at once. If connectionsare
-       timing out and this value is high it indicates either connections storm
-       or unusually slow processing.
-   * - scylla_transport_connections_shed
-     - Holds an incrementing counter with the CQL connections that were shed
-       due to concurrency semaphore timeout (threshold configured via
-       uninitialized_connections_semaphore_cpu_concurrency). This typically can
-       happen during connection.
-   
-  
-
-
-
-
-

docs/upgrade/upgrade-guides/upgrade-guide-from-2025.2-to-2025.3/index.rst

@@ -0,0 +1,13 @@
+==========================================================
+Upgrade - ScyllaDB 2025.2 to ScyllaDB 2025.3
+==========================================================
+
+.. toctree::
+   :maxdepth: 2
+   :hidden:
+
+   Upgrade ScyllaDB <upgrade-guide-from-2025.2-to-2025.3>
+   Metrics Update <metric-update-2025.2-to-2025.3>
+
+* :doc:`Upgrade from ScyllaDB 2025.2.x to ScyllaDB 2025.3.y <upgrade-guide-from-2025.2-to-2025.3>`
+* :doc:`Metrics Update Between 2025.2 and 2025.3 <metric-update-2025.2-to-2025.3>`

docs/upgrade/upgrade-guides/upgrade-guide-from-2025.2-to-2025.3/metric-update-2025.2-to-2025.3.rst

@@ -0,0 +1,95 @@
+.. |SRC_VERSION| replace:: 2025.2
+.. |NEW_VERSION| replace:: 2025.3
+
+================================================================
+Metrics Update Between |SRC_VERSION| and |NEW_VERSION|
+================================================================
+
+.. toctree::
+   :maxdepth: 2
+   :hidden:
+
+ScyllaDB |NEW_VERSION| Dashboards are available as part of the latest |mon_root|.
+
+New Metrics
+------------
+
+The following metrics are new in ScyllaDB |NEW_VERSION| compared to |SRC_VERSION|.
+
+Alternator Per-table Metrics
+===================================
+
+.. list-table::
+   :widths: 25 150
+   :header-rows: 1
+
+   * - Metric
+     - Description
+   * - scylla_alternator_table_batch_item_count
+     - The total number of items processed across all batches.
+   * - scylla_alternator_table_batch_item_count_histogram	
+     - A histogram of the number of items in a batch request.
+   * - scylla_alternator_table_filtered_rows_dropped_total	
+     - The number of rows read and dropped during filtering operations.
+   * - scylla_alternator_table_filtered_rows_matched_total	
+     - The number of rows read and matched during filtering operations.
+   * - scylla_alternator_table_filtered_rows_read_total	
+     - The number of rows read during filtering operations.
+   * - scylla_alternator_table_op_latency
+     - A latency histogram of an operation via Alternator API.
+   * - scylla_alternator_table_op_latency_summary	
+     - A latency summary of an operation via Alternator API.
+   * - scylla_alternator_table_operation
+     - The number of operations via Alternator API.
+   * - scylla_alternator_table_rcu_total
+     - The total number of consumed read units.
+   * - scylla_alternator_table_reads_before_write
+     - The number of performed read-before-write operations.
+   * - scylla_alternator_table_requests_blocked_memory
+     - Counts the number of requests blocked due to memory pressure.
+   * - scylla_alternator_table_requests_shed
+     - Counts the number of requests shed due to overload.
+   * - scylla_alternator_table_shard_bounce_for_lwt
+     - The number of writes that had to be bounced from this shard because of LWT requirements.
+   * - scylla_alternator_table_total_operations
+     - The number of total operations via Alternator API.
+   * - scylla_alternator_table_unsupported_operations
+     - The number of unsupported operations via Alternator API.
+   * - scylla_alternator_table_wcu_total	
+     - The total number of consumed write units.
+   * - scylla_alternator_table_write_using_lwt
+     - The number of writes that used LWT.
+
+Other Metrics
+===============
+
+.. list-table::
+   :widths: 25 150
+   :header-rows: 1
+
+   * - Metric
+     - Description
+   * - scylla_batchlog_manager_total_write_replay_attempts
+     - Counts write operations issued in a batchlog replay flow.
+       A high value of this metric indicates that there is a long batch replay list.
+   * - scylla_corrupt_data_entries_reported	
+     - Counts the number of corrupt data instances reported to the corrupt data handler.
+       A non-zero value indicates that the database suffered data corruption.
+   * - scylla_memory_oversized_allocs
+     - The total count of oversized memory allocations.
+   * - scylla_reactor_internal_errors
+     - The total number of internal errors (subset of cpp_exceptions) that usually
+       indicate a malfunction in the code
+   * - scylla_stall_detector_io_threaded_fallbacks	
+     - The total number of io-threaded-fallbacks operations.
+
+Removed Metrics
+---------------------
+
+The following metrics have been removed in 2025.3:
+
+* scylla_cql_authorized_prepared_statements_cache_evictions
+* scylla_lsa_large_objects_total_space_bytes
+* scylla_lsa_small_objects_total_space_bytes
+* scylla_lsa_small_objects_used_space_bytes
+

docs/upgrade/upgrade-guides/upgrade-guide-from-2025.2-to-2025.3/upgrade-guide-from-2025.2-to-2025.3.rst

@@ -1,13 +1,13 @@
 .. |SCYLLA_NAME| replace:: ScyllaDB
 
-.. |SRC_VERSION| replace:: 2025.1
-.. |NEW_VERSION| replace:: 2025.2
+.. |SRC_VERSION| replace:: 2025.2
+.. |NEW_VERSION| replace:: 2025.3
 
 .. |ROLLBACK| replace:: rollback
 .. _ROLLBACK: ./#rollback-procedure
 
-.. |SCYLLA_METRICS| replace:: ScyllaDB Metrics Update - ScyllaDB 2025.1 to 2025.2
-.. _SCYLLA_METRICS: ../metric-update-2025.1-to-2025.2
+.. |SCYLLA_METRICS| replace:: ScyllaDB Metrics Update - ScyllaDB 2025.2 to 2025.3
+.. _SCYLLA_METRICS: ../metric-update-2025.2-to-2025.3
 
 =======================================================================================
 Upgrade from |SCYLLA_NAME| |SRC_VERSION| to |SCYLLA_NAME| |NEW_VERSION|
@@ -44,14 +44,6 @@ We recommend upgrading the Monitoring Stack to the latest version.
 See the ScyllaDB Release Notes for the latest updates. The Release Notes are published 
 at the `ScyllaDB Community Forum <https://forum.scylladb.com/>`_.
 
-.. note::
-
-   If you previously upgraded from 2024.x to 2025.1 without enabling consistent
-   topology updates, ensure you enable the feature before you upgrade to 2025.2.
-   For instructions, see
-   `Enable Consistent Topology Updates <https://docs.scylladb.com/manual/branch-2025.1/upgrade/upgrade-guides/upgrade-guide-from-2024.x-to-2025.1/enable-consistent-topology.html>`_
-   in the upgrade guide for version 2025.1.
-
 Upgrade Procedure
 =================
 
@@ -158,7 +150,7 @@ You should take note of the current version in case you want to |ROLLBACK|_ the
 
             .. code-block:: console
 
-               sudo wget -O /etc/apt/sources.list.d/scylla.list https://downloads.scylladb.com/deb/debian/scylla-2025.2.list
+               sudo wget -O /etc/apt/sources.list.d/scylla.list https://downloads.scylladb.com/deb/debian/scylla-2025.3.list
 
         #. Install the new ScyllaDB version:
 
@@ -176,7 +168,7 @@ You should take note of the current version in case you want to |ROLLBACK|_ the
 
             .. code-block:: console
 
-               sudo curl -o /etc/yum.repos.d/scylla.repo -L https://downloads.scylladb.com/rpm/centos/scylla-2025.2.repo
+               sudo curl -o /etc/yum.repos.d/scylla.repo -L https://downloads.scylladb.com/rpm/centos/scylla-2025.3.repo
 
         #. Install the new ScyllaDB version:
 

docs/using-scylla/drivers/cql-drivers/index.rst

@@ -16,42 +16,99 @@ ScyllaDB CQL Drivers
 ScyllaDB Drivers
 -----------------
 
+The following ScyllaDB drivers are available:
+
+* :doc:`Python Driver</using-scylla/drivers/cql-drivers/scylla-python-driver>`
+* :doc:`Java Driver </using-scylla/drivers/cql-drivers/scylla-java-driver>`
+* :doc:`Go Driver </using-scylla/drivers/cql-drivers/scylla-go-driver>`
+* :doc:`Go Extension </using-scylla/drivers/cql-drivers/scylla-gocqlx-driver>`
+* :doc:`C++ Driver </using-scylla/drivers/cql-drivers/scylla-cpp-driver>`
+* `CPP-over-Rust Driver <https://github.com/scylladb/cpp-rust-driver>`_
+* :doc:`Rust Driver </using-scylla/drivers/cql-drivers/scylla-rust-driver>`
+
 We recommend using ScyllaDB drivers. All ScyllaDB drivers are shard-aware and provide additional 
 benefits over third-party drivers.
 
 ScyllaDB supports the CQL binary protocol version 3, so any Apache Cassandra/CQL driver that implements 
 the same version works with ScyllaDB.
 
-The following table lists the available ScyllaDB drivers, specifying which support
-`ScyllaDB Cloud Serversless <https://cloud.docs.scylladb.com/stable/serverless/index.html>`_ 
-or include a library for :doc:`CDC </features/cdc/cdc-intro>`.
+CDC Integration with ScyllaDB Drivers
+-------------------------------------------
+
+The following table specifies which ScyllaDB drivers include a library for
+:doc:`CDC </features/cdc/cdc-intro>`.
 
 .. list-table:: 
-   :widths: 30 35 35 
+   :widths: 40 60
    :header-rows: 1
 
-   * - 
-     - ScyllaDB Driver
+   * - ScyllaDB Driver
      - CDC Connector
-   * - :doc:`Python</using-scylla/drivers/cql-drivers/scylla-python-driver>`
-     - |v| 
+   * - :doc:`Python </using-scylla/drivers/cql-drivers/scylla-python-driver>`
      - |x| 
    * - :doc:`Java </using-scylla/drivers/cql-drivers/scylla-java-driver>`
-     - |v| 
      - |v|
    * - :doc:`Go </using-scylla/drivers/cql-drivers/scylla-go-driver>`
-     - |v| 
      - |v|
    * - :doc:`Go Extension </using-scylla/drivers/cql-drivers/scylla-gocqlx-driver>`
-     - |v|
      - |x| 
    * - :doc:`C++ </using-scylla/drivers/cql-drivers/scylla-cpp-driver>`
-     - |v|
      - |x| 
+   * - `CPP-over-Rust Driver <https://github.com/scylladb/cpp-rust-driver>`_
+     - |x|
    * - :doc:`Rust </using-scylla/drivers/cql-drivers/scylla-rust-driver>`
      - |v| 
-     - |v| 
 
+Support for Tablets
+-------------------------
+
+The following table specifies which ScyllaDB drivers support
+:doc:`tablets </architecture/tablets>` and since which version.
+
+.. list-table:: 
+   :widths: 30 35 35
+   :header-rows: 1
+
+   * - ScyllaDB Driver
+     - Support for Tablets
+     - Since Version
+   * - :doc:`Python</using-scylla/drivers/cql-drivers/scylla-python-driver>`
+     - |v| 
+     - 3.26.5
+   * - :doc:`Java </using-scylla/drivers/cql-drivers/scylla-java-driver>`
+     - |v| 
+     - 4.18.0 (Java Driver 4.x)
+
+       3.11.5.2 (Java Driver 3.x)
+   * - :doc:`Go </using-scylla/drivers/cql-drivers/scylla-go-driver>`
+     - |v|
+     - 1.13.0
+   * - :doc:`Go Extension </using-scylla/drivers/cql-drivers/scylla-gocqlx-driver>`
+     - |x| 
+     - N/A
+   * - :doc:`C++ </using-scylla/drivers/cql-drivers/scylla-cpp-driver>`
+     - |x| 
+     - N/A
+   * - `CPP-over-Rust Driver <https://github.com/scylladb/cpp-rust-driver>`_
+     - |v|
+     - All versions
+   * - :doc:`Rust </using-scylla/drivers/cql-drivers/scylla-rust-driver>`
+     - |v| 
+     - 0.13.0
+
+Driver Support Policy
+-------------------------------
+
+We support the **two most recent minor releases** of our drivers.
+
+* We test and validate the latest two minor versions.
+* We typically patch only the latest minor release.
+
+We recommend staying up to date with the latest supported versions to receive
+updates and fixes.
+
+At a minimum, upgrade your driver when upgrading to a new ScyllaDB version
+to ensure compatibility between the driver and the database.
 
 Third-party Drivers
 ----------------------

ent/encryption/encrypted_file_impl.cc

@@ -701,5 +701,97 @@ std::unique_ptr<data_sink_impl> make_encrypted_sink(data_sink sink, shared_ptr<s
     return std::make_unique<encrypted_data_sink>(std::move(sink), std::move(k));
 }
 
+class encrypted_data_source : public data_source_impl, public block_encryption_base {
+    input_stream<char> _input;
+    temporary_buffer<char> _next;
+    size_t _current_position = 0;
+    size_t _skip = 0;
+
+public:
+    encrypted_data_source(data_source source, shared_ptr<symmetric_key> k) 
+        : block_encryption_base(std::move(k))
+        , _input(std::move(source)) 
+    {}
+
+    future<temporary_buffer<char>> get() override {
+        // First, get as much as we can get now (or the remainder of previous call)
+        auto buf1 = _next.empty()
+            ? co_await _input.read()
+            : std::exchange(_next, {})
+            ;
+
+        // eof?
+        if (buf1.empty()) {
+            co_return buf1;
+        }
+
+        // now we need one page more to be able to save one for next lap
+        auto fill_size = align_up(buf1.size(), block_size) + block_size - buf1.size();
+        auto buf2 = co_await _input.read_exactly(fill_size);
+
+        temporary_buffer<char> output(buf1.size() + buf2.size());
+
+        // we copy data even for the part we will cache. this to
+        // fix block alignment of the resulting shared buffer.
+        std::copy(buf1.begin(), buf1.end(), output.get_write());
+        std::copy(buf2.begin(), buf2.end(), output.get_write() + buf1.size());
+
+        // we need to keep one page buffered (beyond the input stream buffer - would be neat 
+        // to share it), to be able to detect actual eof stream size. We always need
+        // at least _two_ pages of data to process, to be able to handle the case where
+        // actual size is <aligned block size> - <less than key block size>.
+        // I.e. stream size = 8180, encrypted data size will be 8192, and data stream
+        // will be 8196. So we need to make sure buf1 == [4096-8192], and buf2 == [8192-8196].
+        if (is_aligned(output.size(), block_size) && output.size() >= 2*block_size) {
+            _next = output.share(output.size() - block_size, block_size);
+            output.trim(output.size() - block_size);
+        }
+
+        const size_t key_block_size = _key->block_size();
+
+        // decrypt all blocks we have to return. might include the last, partial block
+        for (size_t offset = 0; offset < output.size(); offset += block_size, _current_position += block_size) {
+            auto iv = iv_for(_current_position);
+            auto rem = std::min(block_size, output.size() - offset);
+            _key->transform_unpadded(mode::decrypt, output.get() + offset, align_down(rem, key_block_size), output.get_write() + offset, iv.data());
+        }
+
+        // now, if the output buffer is not aligned, we are at eof, and
+        // also need to trim result.
+        if (!is_aligned(output.size(), key_block_size)) {
+            output.trim(output.size() - std::min(output.size(), key_block_size));
+        }
+
+        assert(is_aligned(_current_position, block_size));
+        // finally trim front to handle any skip remainders
+        output.trim_front(std::min(std::exchange(_skip, 0), output.size()));
+
+        co_return output;
+    }
+
+    future<temporary_buffer<char>> skip(uint64_t n) override {
+        if (n >= block_size) {
+            // since we only give back data aligned to block_size chunks,
+            // a client would only ever skip from a block boundary.
+            auto to_skip = align_down(n, block_size);
+            assert(is_aligned(_next.size(), block_size));
+            co_await _input.skip(to_skip - _next.size());
+            n -= to_skip;
+            _current_position += to_skip;
+            _next = {};
+        }
+        _skip = n;
+        co_return temporary_buffer<char>{};
+    }
+
+    future<> close() override { 
+        return _input.close(); 
+    }
+};
+
+
+std::unique_ptr<data_source_impl> make_encrypted_source(data_source source, shared_ptr<symmetric_key> k) {
+    return std::make_unique<encrypted_data_source>(std::move(source), std::move(k));
+}
 }
 

ent/encryption/encrypted_file_impl.hh

@@ -25,4 +25,6 @@ shared_ptr<file_impl> make_delayed_encrypted_file(file, size_t, get_key_func);
 
 std::unique_ptr<data_sink_impl> make_encrypted_sink(data_sink, ::shared_ptr<symmetric_key>);
 
+std::unique_ptr<data_source_impl> make_encrypted_source(data_source source, shared_ptr<symmetric_key> k);
+
 }

ent/encryption/encryption.cc

@@ -472,6 +472,14 @@ public:
             for (auto&& [id, h] : _per_thread_kmip_host_cache[this_shard_id()]) {
                 co_await h->disconnect();
             }
+            static auto stop_all = [](auto&& cache) -> future<> {
+                for (auto& [k, host] : cache) {
+                    co_await host->stop();
+                }
+            };
+            co_await stop_all(_per_thread_kms_host_cache[this_shard_id()]);
+            co_await stop_all(_per_thread_gcp_host_cache[this_shard_id()]);
+
             _per_thread_provider_cache[this_shard_id()].clear();
             _per_thread_system_key_cache[this_shard_id()].clear();
             _per_thread_kmip_host_cache[this_shard_id()].clear();
@@ -676,6 +684,33 @@ public:
         return res;
     }
 
+    std::tuple<opt_bytes, shared_ptr<encryption_schema_extension>> get_encryption_schema_extension(const sstables::sstable& sst,
+                                                                                                   sstables::component_type type) const {
+        const auto& sc = sst.get_shared_components();
+        if (!sc.scylla_metadata) {
+            return {};
+        }
+        const auto* ext_attr = sc.scylla_metadata->get_extension_attributes();
+        if (!ext_attr) {
+            return {};
+        }
+
+        bool ok = ext_attr->map.contains(encryption_attribute_ds);
+        if (ok && type != sstables::component_type::Data) {
+            ok = (ser::deserialize_from_buffer(ext_attr->map.at(encrypted_components_attribute_ds).value, std::type_identity<uint32_t>{}, 0) & (1 << static_cast<int>(type))) > 0;
+        }
+
+        if (!ok) {
+            return {};
+        }
+        auto esx = encryption_schema_extension::create(*_ctxt, ext_attr->map.at(encryption_attribute_ds).value);
+        opt_bytes id;
+        if (ext_attr->map.contains(key_id_attribute_ds)) {
+            id = ext_attr->map.at(key_id_attribute_ds).value;
+        }
+        return {std::move(id), std::move(esx)};
+    }
+
     future<file> wrap_file(const sstables::sstable& sst, sstables::component_type type, file f, open_flags flags) override {
         switch (type) {
         case sstables::component_type::Scylla:
@@ -688,44 +723,21 @@ public:
 
         if (flags == open_flags::ro) {
             // open existing. check read opts.
-            auto& sc = sst.get_shared_components();
-            if (sc.scylla_metadata) {
-                auto* exta  = sc.scylla_metadata->get_extension_attributes();
-                if (exta) {
-                    auto i = exta->map.find(encryption_attribute_ds);
-                    // note: earlier builds of encryption extension would only encrypt data component,
-                    // so iff we are opening old sstables we need to check if this component is actually
-                    // encrypted. We use a bitmask attribute for this.
+            auto [id, esx] = get_encryption_schema_extension(sst, type);
+            if (esx) {
+                if (esx->should_delay_read(id)) {
+                    logg.debug("Encrypted sstable component {} using delayed opening {} (id: {})", sst.component_basename(type), *esx, id);
 
-                    bool ok = i != exta->map.end();
-                    if (ok && type != sstables::component_type::Data) {
-                        ok = exta->map.count(encrypted_components_attribute_ds) &&
-                                        (ser::deserialize_from_buffer(exta->map.at(encrypted_components_attribute_ds).value, std::type_identity<uint32_t>{}, 0) & (1 << int(type)));
-                    }
-
-                    if (ok) {
-                        auto esx = encryption_schema_extension::create(*_ctxt, i->second.value);
-                        opt_bytes id;
-
-                        if (exta->map.count(key_id_attribute_ds)) {
-                            id = exta->map.at(key_id_attribute_ds).value;
-                        }
-
-                        if (esx->should_delay_read(id)) {
-                            logg.debug("Encrypted sstable component {} using delayed opening {} (id: {})", sst.component_basename(type), *esx, id);
-
-                            co_return make_delayed_encrypted_file(f, esx->key_block_size(), [esx, comp = sst.component_basename(type), id = std::move(id)] {
-                                logg.trace("Delayed component {} using {} (id: {}) resolve", comp, *esx, id);
-                                return esx->key_for_read(id);
-                            });
-                        }
-
-                        logg.debug("Open encrypted sstable component {} using {} (id: {})", sst.component_basename(type), *esx, id);
-
-                        auto k = co_await esx->key_for_read(std::move(id));
-                        co_return make_encrypted_file(f, std::move(k));
-                    }
+                    co_return make_delayed_encrypted_file(f, esx->key_block_size(), [esx, comp = sst.component_basename(type), id = std::move(id)] {
+                        logg.trace("Delayed component {} using {} (id: {}) resolve", comp, *esx, id);
+                        return esx->key_for_read(id);
+                    });
                 }
+
+                logg.debug("Open encrypted sstable component {} using {} (id: {})", sst.component_basename(type), *esx, id);
+
+                auto k = co_await esx->key_for_read(std::move(id));
+                co_return make_encrypted_file(f, std::move(k));
             }
         } else {
             if (co_await wrap_writeonly(sst, type, [&f](shared_ptr<symmetric_key> k) { f = make_encrypted_file(std::move(f), std::move(k)); })) {
@@ -823,6 +835,36 @@ public:
         });
         co_return sink;
     }
+
+    future<data_source> wrap_source(const sstables::sstable& sst,
+                                                         sstables::component_type type,
+                                                         sstables::data_source_creator_fn data_source_creator,
+                                                         uint64_t offset,
+                                                         uint64_t len) override {
+        switch (type) {
+        case sstables::component_type::Scylla:
+        case sstables::component_type::TemporaryTOC:
+        case sstables::component_type::TOC:
+            co_return data_source_creator(offset, len);
+        case sstables::component_type::CompressionInfo:
+        case sstables::component_type::CRC:
+        case sstables::component_type::Data:
+        case sstables::component_type::Digest:
+        case sstables::component_type::Filter:
+        case sstables::component_type::Index:
+        case sstables::component_type::Statistics:
+        case sstables::component_type::Summary:
+        case sstables::component_type::TemporaryStatistics:
+        case sstables::component_type::Unknown:
+            auto [id, esx] = get_encryption_schema_extension(sst, type);
+            if (esx) {
+                auto key = co_await esx->key_for_read(std::move(id));
+                auto block_size = key->block_size();
+                co_return data_source(make_encrypted_source(data_source_creator(align_down(offset, block_size), align_up(len, block_size)), std::move(key)));
+            }
+            co_return data_source_creator(offset, len);
+        }
+    }
 };
 
 std::string encryption_provider(const sstables::sstable& sst) {

ent/encryption/gcp_host.cc

@@ -97,6 +97,7 @@ public:
     ~impl() = default;
 
     future<> init();
+    future<> stop();
     const host_options& options() const {
         return _options;
     }
@@ -477,13 +478,14 @@ encryption::gcp_host::impl::get_default_credentials() {
         }
     }
 
-    {
+    auto home = std::getenv("HOME");
+    if (home) {
         std::string well_known_file;
         auto env_path = std::getenv("CLOUDSDK_CONFIG");
         if (env_path) {
-            well_known_file = fmt::format("~/{}/{}", env_path, WELL_KNOWN_CREDENTIALS_FILE);
+            well_known_file = fmt::format("{}/{}/{}", home, env_path, WELL_KNOWN_CREDENTIALS_FILE);
         } else {
-            well_known_file = fmt::format("~/.config/{}/{}", CLOUDSDK_CONFIG_DIRECTORY, WELL_KNOWN_CREDENTIALS_FILE);
+            well_known_file = fmt::format("{}/.config/{}/{}", home, CLOUDSDK_CONFIG_DIRECTORY, WELL_KNOWN_CREDENTIALS_FILE);
         }
 
         if (co_await seastar::file_exists(well_known_file)) {
@@ -671,7 +673,7 @@ encryption::gcp_host::impl::get_access_token(const google_credentials& creds, co
                 { "client_id", c.client_id },
                 { "client_secret", c.client_secret },
                 { "refresh_token", c.refresh_token },
-                { "grant_type", "grant_type" },
+                { "grant_type", "refresh_token" },
             }), "", httpd::operation_type::POST);
 
             co_return access_token{ json };
@@ -827,6 +829,11 @@ future<> encryption::gcp_host::impl::init() {
     _initialized = true;
 }
 
+future<> encryption::gcp_host::impl::stop() {
+    co_await _attr_cache.stop();
+    co_await _id_cache.stop();
+}
+
 std::tuple<std::string, std::string> encryption::gcp_host::impl::parse_key(std::string_view spec) {
     auto i = spec.find_last_of('/');
     if (i == std::string_view::npos) {
@@ -989,6 +996,10 @@ future<> encryption::gcp_host::init() {
     return _impl->init();
 }
 
+future<> encryption::gcp_host::stop() {
+    return _impl->stop();
+}
+
 const encryption::gcp_host::host_options& encryption::gcp_host::options() const {
     return _impl->options();
 }

ent/encryption/gcp_host.hh

@@ -65,6 +65,8 @@ public:
     ~gcp_host();
 
     future<> init();
+    future<> stop();
+
     const host_options& options() const;
 
     struct option_override : public t_credentials_source<std::optional<std::string>> {

ent/encryption/kmip_host.cc

@@ -724,9 +724,11 @@ future<> kmip_host::impl::connect() {
 }
 
 future<> kmip_host::impl::disconnect() {
-    return do_for_each(_options.hosts, [this](const sstring& host) {
+    co_await do_for_each(_options.hosts, [this](const sstring& host) {
         return clear_connections(host);
     });
+    co_await _attr_cache.stop();
+    co_await _id_cache.stop();
 }
 
 static unsigned from_str(unsigned (*f)(char*, int, int*), const sstring& s, const sstring& what) {

ent/encryption/kms_host.cc

@@ -154,6 +154,8 @@ public:
     ~impl() = default;
 
     future<> init();
+    future<> stop();
+
     const host_options& options() const {
         return _options;
     }
@@ -826,6 +828,11 @@ future<> encryption::kms_host::impl::init() {
     _initialized = true;
 }
 
+future<> encryption::kms_host::impl::stop() {
+    co_await _attr_cache.stop();
+    co_await _id_cache.stop();
+}
+
 future<encryption::kms_host::impl::key_and_id_type> encryption::kms_host::impl::create_key(const attr_cache_key& k) {
     auto& master_key = k.master_key;
     auto& aws_assume_role_arn = k.aws_assume_role_arn;
@@ -988,6 +995,10 @@ future<> encryption::kms_host::init() {
     return _impl->init();
 }
 
+future<> encryption::kms_host::stop() {
+    return _impl->stop();
+}
+
 const encryption::kms_host::host_options& encryption::kms_host::options() const {
     return _impl->options();
 }

ent/encryption/kms_host.hh

@@ -63,6 +63,8 @@ public:
     ~kms_host();
 
     future<> init();
+    future<> stop();
+
     const host_options& options() const;
 
     struct option_override {

gms/gossip_digest_syn.cc

@@ -13,8 +13,8 @@
 auto fmt::formatter<gms::gossip_digest_syn>::format(const gms::gossip_digest_syn& syn, fmt::format_context& ctx) const
         -> decltype(ctx.out()) {
     auto out = ctx.out();
-    out = fmt::format_to(out, "cluster_id:{},partioner:{},group0_id{},",
-                         syn._cluster_id, syn._partioner, syn._group0_id);
+    out = fmt::format_to(out, "cluster_id:{},partioner:{},group0_id:{},recovery_leader:{}",
+                         syn._cluster_id, syn._partioner, syn._group0_id, syn._recovery_leader);
     out = fmt::format_to(out, "digests:{{");
     for (auto& d : syn._digests) {
         out = fmt::format_to(out, "{} ", d);

gms/gossip_digest_syn.hh

@@ -28,15 +28,19 @@ private:
     sstring _partioner;
     utils::chunked_vector<gossip_digest> _digests;
     utils::UUID _group0_id;
+    utils::UUID _recovery_leader;
 public:
     gossip_digest_syn() {
     }
 
-    gossip_digest_syn(sstring id, sstring p, utils::chunked_vector<gossip_digest> digests, utils::UUID group0_id)
+    gossip_digest_syn(
+            sstring id, sstring p, utils::chunked_vector<gossip_digest> digests,
+            utils::UUID group0_id, utils::UUID recovery_leader)
         : _cluster_id(std::move(id))
         , _partioner(std::move(p))
         , _digests(std::move(digests))
-        , _group0_id(std::move(group0_id)) {
+        , _group0_id(std::move(group0_id))
+        , _recovery_leader(std::move(recovery_leader)) {
     }
 
     sstring cluster_id() const {
@@ -47,6 +51,10 @@ public:
         return _group0_id;
     }
 
+    utils::UUID recovery_leader() const {
+        return _recovery_leader;
+    }
+
     sstring partioner() const {
         return _partioner;
     }
@@ -59,6 +67,10 @@ public:
         return group0_id();
     }
 
+    utils::UUID get_recovery_leader() const {
+        return _recovery_leader;
+    }
+
     sstring get_partioner() const {
         return partioner();
     }

gms/gossiper.cc

@@ -76,6 +76,10 @@ const utils::UUID& gossiper::get_group0_id() const noexcept {
     return _gcfg.group0_id;
 }
 
+const utils::UUID& gossiper::get_recovery_leader() const noexcept {
+    return _gcfg.recovery_leader;
+}
+
 const std::set<inet_address>& gossiper::get_seeds() const noexcept {
     return _gcfg.seeds;
 }
@@ -172,8 +176,11 @@ void gossiper::do_sort(utils::chunked_vector<gossip_digest>& g_digest_list) cons
 // Depends on
 // - no external dependency
 future<> gossiper::handle_syn_msg(locator::host_id from, gossip_digest_syn syn_msg) {
-    logger.trace("handle_syn_msg():from={},cluster_name:peer={},local={},group0_id:peer={},local={},partitioner_name:peer={},local={}",
-        from, syn_msg.cluster_id(), get_cluster_name(), syn_msg.group0_id(), get_group0_id(), syn_msg.partioner(), get_partitioner_name());
+    logger.trace(
+            "handle_syn_msg():from={},cluster_name:peer={},local={},group0_id:peer={},local={},"
+            "recovery_leader:peer={},local={},partitioner_name:peer={},local={}",
+            from, syn_msg.cluster_id(), get_cluster_name(), syn_msg.group0_id(), get_group0_id(),
+            syn_msg.recovery_leader(), get_recovery_leader(), syn_msg.partioner(), get_partitioner_name());
     if (!is_enabled()) {
         co_return;
     }
@@ -184,10 +191,20 @@ future<> gossiper::handle_syn_msg(locator::host_id from, gossip_digest_syn syn_m
         co_return;
     }
 
+    // Recovery leader mismatch implies an administrator's mistake during the Raft-based recovery procedure.
+    // Throw away the message and signal that something is wrong.
+    bool both_nodes_in_recovery = syn_msg.recovery_leader() && get_recovery_leader();
+    if (both_nodes_in_recovery && syn_msg.recovery_leader() != get_recovery_leader()) {
+        logger.warn("Recovery leader mismatch from {} {} != {},",
+                from, syn_msg.recovery_leader(), get_recovery_leader());
+        co_return;
+    }
+
     // If the message is from a node with a different group0 id throw it away.
     // A group0 id mismatch is expected during a rolling restart in the Raft-based recovery procedure.
-    if (_gcfg.recovery_leader().empty()
-            && syn_msg.group0_id() && get_group0_id() && syn_msg.group0_id() != get_group0_id()) {
+    bool no_recovery = !syn_msg.recovery_leader() && !get_recovery_leader();
+    bool group0_ids_mismatch = syn_msg.group0_id() && get_group0_id() && syn_msg.group0_id() != get_group0_id();
+    if (no_recovery && group0_ids_mismatch) {
         logger.warn("Group0Id mismatch from {} {} != {}", from, syn_msg.group0_id(), get_group0_id());
         co_return;
     }
@@ -1102,7 +1119,8 @@ void gossiper::run() {
             utils::chunked_vector<gossip_digest> g_digests = make_random_gossip_digest();
 
             if (g_digests.size() > 0) {
-                gossip_digest_syn message(get_cluster_name(), get_partitioner_name(), g_digests, get_group0_id());
+                gossip_digest_syn message(
+                        get_cluster_name(), get_partitioner_name(), g_digests, get_group0_id(), get_recovery_leader());
 
                 if (_endpoints_to_talk_with.empty() && !_live_endpoints.empty()) {
                     auto live_endpoints = _live_endpoints | std::ranges::to<std::vector>();

gms/gossiper.hh

@@ -69,7 +69,7 @@ struct gossip_config {
     uint32_t skip_wait_for_gossip_to_settle = -1;
     utils::updateable_value<uint32_t> failure_detector_timeout_ms;
     utils::updateable_value<int32_t> force_gossip_generation;
-    utils::updateable_value<sstring> recovery_leader;
+    utils::updateable_value<utils::UUID> recovery_leader;
 };
 
 struct loaded_endpoint_state {
@@ -136,6 +136,8 @@ public:
     void set_group0_id(utils::UUID group0_id);
     const utils::UUID& get_group0_id() const noexcept;
 
+    const utils::UUID& get_recovery_leader() const noexcept;
+
     const sstring& get_partitioner_name() const noexcept {
         return _gcfg.partitioner;
     }

idl/gossip_digest.idl.hh

@@ -57,6 +57,7 @@ class gossip_digest_syn {
     sstring get_partioner();
     utils::chunked_vector<gms::gossip_digest> get_gossip_digests();
     utils::UUID get_group0_id()[[version 5.4]];
+    utils::UUID get_recovery_leader()[[version 2025.2.2]];
 };
 
 class gossip_digest_ack {

locator/tablets.cc

@@ -335,18 +335,25 @@ void tablet_metadata::drop_tablet_map(table_id id) {
 }
 
 future<> tablet_metadata::clear_gently() {
-    for (auto&& [id, map] : _tablets) {
-        const auto shard = map.get_owner_shard();
-        co_await smp::submit_to(shard, [map = std::move(map)] () mutable {
-            auto map_ptr = map.release();
-            // Others copies exist, we simply drop ours, no need to clear anything.
-            if (map_ptr.use_count() > 1) {
-                return make_ready_future<>();
-            }
-            return const_cast<tablet_map&>(*map_ptr).clear_gently().finally([map_ptr = std::move(map_ptr)] { });
-        });
+    tablet_logger.debug("tablet_metadata::clear_gently {}", fmt::ptr(this));
+    // First, Sort the tablet maps per shard to avoid destruction of all foreign tablet map ptrs
+    // on this shard. We don't use sharded<> here since it will require a similar 
+    // submit_to to each shard owner per tablet-map.
+    std::vector<std::vector<tablet_map_ptr>> tablet_maps_per_shard;
+    tablet_maps_per_shard.resize(smp::count);
+    for (auto& [_, map_ptr] : _tablets) {
+        tablet_maps_per_shard[map_ptr.get_owner_shard()].emplace_back(std::move(map_ptr));
     }
     _tablets.clear();
+
+    // Now destroy the foreign tablet map pointers on each shard.
+    co_await smp::invoke_on_all([&] -> future<> {
+        for (auto& map_ptr : tablet_maps_per_shard[this_shard_id()]) {
+            auto map = map_ptr.release();
+            co_await utils::clear_gently(map);
+        }
+    });
+
     co_return;
 }
 

locator/token_metadata.cc

@@ -357,6 +357,7 @@ future<std::unique_ptr<token_metadata_impl>> token_metadata_impl::clone_only_tok
 }
 
 future<> token_metadata_impl::clear_gently() noexcept {
+    _version_tracker = {};
     co_await utils::clear_gently(_token_to_endpoint_map);
     co_await utils::clear_gently(_normal_token_owners);
     co_await utils::clear_gently(_bootstrap_tokens);
@@ -834,16 +835,30 @@ token_metadata::token_metadata(std::unique_ptr<token_metadata_impl> impl)
 {
 }
 
-token_metadata::token_metadata(config cfg)
-        : _impl(std::make_unique<token_metadata_impl>(cfg))
+token_metadata::token_metadata(shared_token_metadata& stm, config cfg)
+        : _shared_token_metadata(&stm)
+        , _impl(std::make_unique<token_metadata_impl>(std::move(cfg)))
 {
 }
 
-token_metadata::~token_metadata() = default;
+token_metadata::~token_metadata() {
+    clear_and_dispose_impl();
+}
 
 token_metadata::token_metadata(token_metadata&&) noexcept = default;
 
-token_metadata& token_metadata::token_metadata::operator=(token_metadata&&) noexcept = default;
+token_metadata& token_metadata::token_metadata::operator=(token_metadata&& o) noexcept {
+    if (this != &o) {
+        clear_and_dispose_impl();
+        _shared_token_metadata = std::exchange(o._shared_token_metadata, nullptr);
+        _impl = std::exchange(o._impl, nullptr);
+    }
+    return *this;
+}
+
+void token_metadata::set_shared_token_metadata(shared_token_metadata& stm) {
+    _shared_token_metadata = &stm;
+}
 
 const std::vector<token>&
 token_metadata::sorted_tokens() const {
@@ -1027,6 +1042,15 @@ token_metadata::clone_after_all_left() const noexcept {
     co_return token_metadata(co_await _impl->clone_after_all_left());
 }
 
+void token_metadata::clear_and_dispose_impl() noexcept {
+    if (!_shared_token_metadata) {
+        return;
+    }
+    if (auto impl = std::exchange(_impl, nullptr)) {
+        _shared_token_metadata->clear_and_dispose(std::move(impl));
+    }
+}
+
 future<> token_metadata::clear_gently() noexcept {
     return _impl->clear_gently();
 }
@@ -1143,6 +1167,17 @@ version_tracker shared_token_metadata::new_tracker(token_metadata::version_t ver
     return tracker;
 }
 
+future<> shared_token_metadata::stop() noexcept {
+    co_await _background_dispose_gate.close();
+}
+
+void shared_token_metadata::clear_and_dispose(std::unique_ptr<token_metadata_impl> impl) noexcept {
+    // Safe to drop the future since the gate is closed in stop()
+    if (auto gh = _background_dispose_gate.try_hold()) {
+        (void)impl->clear_gently().finally([i = std::move(impl), gh = std::move(gh)] {});
+    }
+}
+
 void shared_token_metadata::set(mutable_token_metadata_ptr tmptr) noexcept {
     if (_shared->get_ring_version() >= tmptr->get_ring_version()) {
         on_internal_error(tlogger, format("shared_token_metadata: must not set non-increasing ring_version: {} -> {}", _shared->get_ring_version(), tmptr->get_ring_version()));
@@ -1154,6 +1189,7 @@ void shared_token_metadata::set(mutable_token_metadata_ptr tmptr) noexcept {
         _stale_versions_in_use = _versions_barrier.advance_and_await();
     }
 
+    tmptr->set_shared_token_metadata(*this);
     _shared = std::move(tmptr);
     _shared->set_version_tracker(new_tracker(_shared->get_version()));
 
@@ -1216,7 +1252,7 @@ future<> shared_token_metadata::mutate_on_all_shards(sharded<shared_token_metada
 
     std::vector<mutable_token_metadata_ptr> pending_token_metadata_ptr;
     pending_token_metadata_ptr.resize(smp::count);
-    auto tmptr = make_token_metadata_ptr(co_await stm.local().get()->clone_async());
+    auto tmptr = stm.local().make_token_metadata_ptr(co_await stm.local().get()->clone_async());
     auto& tm = *tmptr;
     // bump the token_metadata ring_version
     // to invalidate cached token/replication mappings
@@ -1227,7 +1263,7 @@ future<> shared_token_metadata::mutate_on_all_shards(sharded<shared_token_metada
     // Apply the mutated token_metadata only after successfully cloning it on all shards.
     pending_token_metadata_ptr[base_shard] = tmptr;
     co_await smp::invoke_on_others(base_shard, [&] () -> future<> {
-        pending_token_metadata_ptr[this_shard_id()] = make_token_metadata_ptr(co_await tm.clone_async());
+        pending_token_metadata_ptr[this_shard_id()] = stm.local().make_token_metadata_ptr(co_await tm.clone_async());
     });
 
     co_await stm.invoke_on_all([&] (shared_token_metadata& stm) {

locator/token_metadata.hh

@@ -47,7 +47,7 @@ class abstract_replication_strategy;
 
 using token = dht::token;
 
-class token_metadata;
+class shared_token_metadata;
 class tablet_metadata;
 
 struct host_id_or_endpoint {
@@ -166,6 +166,7 @@ private:
 };
 
 class token_metadata final {
+    shared_token_metadata* _shared_token_metadata = nullptr;
     std::unique_ptr<token_metadata_impl> _impl;
 private:
     friend class token_metadata_ring_splitter;
@@ -178,7 +179,7 @@ public:
     using version_t = service::topology::version_t;
     using version_tracker_t = version_tracker;
 
-    token_metadata(config cfg);
+    token_metadata(shared_token_metadata& stm, config cfg);
     explicit token_metadata(std::unique_ptr<token_metadata_impl> impl);
     token_metadata(token_metadata&&) noexcept; // Can't use "= default;" - hits some static_assert in unique_ptr
     token_metadata& operator=(token_metadata&&) noexcept;
@@ -355,6 +356,11 @@ public:
     friend class shared_token_metadata;
 private:
     void set_version_tracker(version_tracker_t tracker);
+
+    void set_shared_token_metadata(shared_token_metadata& stm);
+
+    // Clears and disposes the token metadata impl in the background, if present.
+    void clear_and_dispose_impl() noexcept;
 };
 
 struct topology_change_info {
@@ -371,12 +377,8 @@ struct topology_change_info {
 using token_metadata_lock = semaphore_units<>;
 using token_metadata_lock_func = noncopyable_function<future<token_metadata_lock>() noexcept>;
 
-template <typename... Args>
-mutable_token_metadata_ptr make_token_metadata_ptr(Args... args) {
-    return make_lw_shared<token_metadata>(std::forward<Args>(args)...);
-}
-
-class shared_token_metadata {
+class shared_token_metadata : public peering_sharded_service<shared_token_metadata> {
+    named_gate _background_dispose_gate{"shared_token_metadata::background_dispose_gate"};
     mutable_token_metadata_ptr _shared;
     token_metadata_lock_func _lock_func;
     std::chrono::steady_clock::duration _stall_detector_threshold = std::chrono::seconds(2);
@@ -408,7 +410,7 @@ public:
     // used to construct the shared object as a sharded<> instance
     // lock_func returns semaphore_units<>
     explicit shared_token_metadata(token_metadata_lock_func lock_func, token_metadata::config cfg)
-        : _shared(make_token_metadata_ptr(std::move(cfg)))
+        : _shared(make_lw_shared<token_metadata>(*this, cfg))
         , _lock_func(std::move(lock_func))
         , _versions_barrier("shared_token_metadata::versions_barrier")
     {
@@ -418,6 +420,17 @@ public:
     shared_token_metadata(const shared_token_metadata& x) = delete;
     shared_token_metadata(shared_token_metadata&& x) = default;
 
+    future<> stop() noexcept;
+
+    mutable_token_metadata_ptr make_token_metadata_ptr() {
+        return make_lw_shared<token_metadata>(*this, token_metadata::config{_shared->get_topology().get_config()});
+    }
+
+    mutable_token_metadata_ptr make_token_metadata_ptr(token_metadata&& tm) {
+        tm.set_shared_token_metadata(*this);
+        return make_lw_shared<token_metadata>(std::move(tm));
+    }
+
     token_metadata_ptr get() const noexcept {
         return _shared;
     }
@@ -467,6 +480,8 @@ public:
     // Must be called on shard 0.
     static future<> mutate_on_all_shards(sharded<shared_token_metadata>& stm, seastar::noncopyable_function<future<> (token_metadata&)> func);
 
+    void clear_and_dispose(std::unique_ptr<token_metadata_impl> impl) noexcept;
+
 private:
     // for testing only, unsafe to be called without awaiting get_lock() first
     void mutate_token_metadata_for_test(seastar::noncopyable_function<void (token_metadata&)> func);

pgo/profiles/aarch64/profile.profdata.xz

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:f69e30ac03713e439d4f9fe347aafe2201d8605880358d3142b6f6bc706c3014
-size 5966816
+oid sha256:e1364fb228ed061377ade0165fd21791eaaab2220bfedbeaa6173207e6bd3e7a
+size 6018840

pgo/profiles/x86_64/profile.profdata.xz

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:9ec68edb2980fae1fcf63046b399f30b882fc7b77b4bc316c7055f75820d26f1
-size 5975376
+oid sha256:1e488c3311ec2a7abcfdb4f330a6f5dc4fb2de98ae4f18a315da7d6de830e314
+size 6040896

repair/repair.cc

@@ -12,6 +12,7 @@
 #include "repair/row_level.hh"
 
 #include "locator/network_topology_strategy.hh"
+#include "seastar/core/on_internal_error.hh"
 #include "streaming/stream_reason.hh"
 #include "gms/inet_address.hh"
 #include "gms/gossiper.hh"
@@ -1452,10 +1453,6 @@ future<std::optional<double>> repair::user_requested_repair_task_impl::expected_
     co_return _ranges.size() * _cfs.size() * smp::count;
 }
 
-std::optional<double> repair::user_requested_repair_task_impl::expected_children_number() const {
-    return smp::count;
-}
-
 future<int> repair_start(seastar::sharded<repair_service>& repair, sharded<gms::gossip_address_map>& am,
         sstring keyspace, std::unordered_map<sstring, sstring> options) {
     return repair.invoke_on(0, [keyspace = std::move(keyspace), options = std::move(options), &am] (repair_service& local_repair) {
@@ -1624,10 +1621,6 @@ future<std::optional<double>> repair::data_sync_repair_task_impl::expected_total
     co_return _cfs_size ? std::make_optional<double>(_ranges.size() * _cfs_size * smp::count) : std::nullopt;
 }
 
-std::optional<double> repair::data_sync_repair_task_impl::expected_children_number() const {
-    return smp::count;
-}
-
 future<> repair_service::bootstrap_with_repair(locator::token_metadata_ptr tmptr, std::unordered_set<dht::token> bootstrap_tokens) {
     SCYLLA_ASSERT(this_shard_id() == 0);
     return seastar::async([this, tmptr = std::move(tmptr), tokens = std::move(bootstrap_tokens)] () mutable {
@@ -2244,7 +2237,7 @@ future<> repair_service::replace_with_repair(std::unordered_map<sstring, locator
     auto reason = streaming::stream_reason::replace;
     // update a cloned version of tmptr
     // no need to set the original version
-    auto cloned_tmptr = make_token_metadata_ptr(std::move(cloned_tm));
+    auto cloned_tmptr = _db.local().get_shared_token_metadata().make_token_metadata_ptr(std::move(cloned_tm));
     cloned_tmptr->update_topology(tmptr->get_my_id(), myloc, locator::node::state::replacing);
     co_await cloned_tmptr->update_normal_tokens(replacing_tokens, tmptr->get_my_id());
     auto source_dc = utils::optional_param(myloc.dc);
@@ -2267,9 +2260,92 @@ static std::unordered_set<locator::host_id> get_token_owners_in_dcs(std::vector<
     return dc_endpoints;
 }
 
+future<> repair::remote_metas::for_each_local_meta(std::function<future<>(const tablet_repair_task_meta&)> func) const {
+    size_t this_shard = this_shard_id();
+    if (!_metas_on_shards[this_shard]) {
+        co_return;
+    }
+    for (size_t i = 0; i < _metas_on_shards[this_shard]->metas.size(); ++i) {
+        co_await func(_metas_on_shards[this_shard]->metas[i]);
+    }
+}
+
+future<repair::hosts_and_tables> repair::remote_metas::get_hosts_and_tables() const {
+    repair::hosts_and_tables ret;
+
+    co_await coroutine::parallel_for_each(smp::all_cpus(), [this, &ret](const auto& shard) -> future<> {
+        auto hosts_and_tables = co_await smp::submit_to(shard, [this]() mutable -> future<repair::hosts_and_tables> {
+            repair::hosts_and_tables shard_ret;
+
+            co_await for_each_local_meta([&shard_ret](const auto& meta) {
+                shard_ret.hosts.insert(meta.neighbors.all.begin(), meta.neighbors.all.end());
+                shard_ret.tables.insert(meta.tid);
+                return make_ready_future<>();
+            });
+
+            co_return shard_ret;
+        });
+
+        ret.hosts.insert(hosts_and_tables.hosts.begin(), hosts_and_tables.hosts.end());
+        ret.tables.insert(hosts_and_tables.tables.begin(), hosts_and_tables.tables.end());
+    });
+
+    co_return ret;
+}
+
+size_t repair::remote_metas::size() const {
+    return _metas_count;
+}
+
+void repair::remote_metas::clear() {
+    _metas_on_shards.clear();
+    _metas_count = 0;
+}
+
+future<> repair::remote_metas_builder::allocate_on_shard(size_t shard_id) {
+    _remote_metas._metas_on_shards[shard_id] = co_await smp::submit_to(shard_id, []() -> future<remote_metas::remote_data_ptr> {
+        auto ptr = make_lw_shared<remote_data>();
+        co_return remote_metas::remote_data_ptr(std::move(ptr));
+    });
+}
+
+future<> repair::remote_metas_builder::flush(size_t shard_id) {
+    auto it = _pending_metas.find(shard_id);
+    if (it == _pending_metas.end() || it->second.empty()) {
+        co_return;
+    }
+    auto local_pending_metas = std::move(it->second);
+    _pending_metas.erase(it);
+
+    auto& metas_ptr = _remote_metas._metas_on_shards[shard_id];
+    if (!metas_ptr) {
+        co_await allocate_on_shard(shard_id);
+    }
+    co_await smp::submit_to(shard_id, [metas = std::move(local_pending_metas), &metas_ptr]() {
+        metas_ptr->metas.insert(metas_ptr->metas.end(), std::make_move_iterator(metas.begin()), std::make_move_iterator(metas.end()));
+    });
+}
+
+future<> repair::remote_metas_builder::add_on_shard(size_t shard_id, tablet_repair_task_meta meta) {
+    ++_remote_metas._metas_count;
+    auto& local_pending_metas = _pending_metas[shard_id];
+    local_pending_metas.push_back(std::move(meta));
+    if (local_pending_metas.size() >= max_pending_metas_per_shard) {
+        co_await flush(shard_id);
+    }
+}
+
+future<repair::remote_metas> repair::remote_metas_builder::build() && {
+    for (size_t shard_id = 0; shard_id < smp::count; ++shard_id) {
+        co_await flush(shard_id);
+    }
+    co_return std::move(_remote_metas);
+}
+
 // Repair all tablets belong to this node for the given table
 future<> repair_service::repair_tablets(repair_uniq_id rid, sstring keyspace_name, std::vector<sstring> table_names, bool primary_replica_only, dht::token_range_vector ranges_specified, std::vector<sstring> data_centers, std::unordered_set<locator::host_id> hosts, std::unordered_set<locator::host_id> ignore_nodes, std::optional<int> ranges_parallelism) {
-    std::vector<tablet_repair_task_meta> task_metas;
+    utils::chunked_vector<locator::effective_replication_map_ptr> erms;
+    repair::remote_metas_builder task_metas_builder;
     for (auto& table_name : table_names) {
         lw_shared_ptr<replica::table> t;
         try {
@@ -2283,7 +2359,8 @@ future<> repair_service::repair_tablets(repair_uniq_id rid, sstring keyspace_nam
         }
         table_id tid = t->schema()->id();
         // Invoke group0 read barrier before obtaining erm pointer so that it sees all prior metadata changes
-        auto dropped = co_await streaming::table_sync_and_check(_db.local(), _mm, tid);
+        auto dropped = !utils::get_local_injector().enter("repair_tablets_no_sync") &&
+            co_await streaming::table_sync_and_check(_db.local(), _mm, tid);
         if (dropped) {
             rlogger.debug("repair[{}] Table {}.{} does not exist anymore", rid.uuid(), keyspace_name, table_name);
             continue;
@@ -2292,11 +2369,15 @@ future<> repair_service::repair_tablets(repair_uniq_id rid, sstring keyspace_nam
         while (true) {
             _repair_module->check_in_shutdown();
             erm = t->get_effective_replication_map();
+            auto local_version = erm->get_token_metadata().get_version();
             const locator::tablet_map& tmap = erm->get_token_metadata_ptr()->tablets().get_tablet_map(tid);
-            if (!tmap.has_transitions()) {
+            if (!tmap.has_transitions() && co_await container().invoke_on(0, [local_version] (repair_service& rs) {
+                    // We need to ensure that there is no ongoing global request.
+                    return local_version == rs._tsm.local()._topology.version && !rs._tsm.local()._topology.is_busy();
+                })) {
                 break;
             }
-            rlogger.info("repair[{}] Table {}.{} has tablet transitions, waiting for topology to quiesce", rid.uuid(), keyspace_name, table_name);
+            rlogger.info("repair[{}] Topology is busy, waiting for it to quiesce", rid.uuid());
             erm = nullptr;
             co_await container().invoke_on(0, [] (repair_service& rs) {
                 return rs._tsm.local().await_not_busy();
@@ -2369,6 +2450,7 @@ future<> repair_service::repair_tablets(repair_uniq_id rid, sstring keyspace_nam
         }
 
         size_t nr = 0;
+        bool metas_added = false;
         for (auto& m : metas) {
             nr++;
             rlogger.debug("repair[{}] Collect {} out of {} tablets: table={}.{} tablet_id={} range={} replicas={} primary_replica_only={}",
@@ -2428,14 +2510,17 @@ future<> repair_service::repair_tablets(repair_uniq_id rid, sstring keyspace_nam
                 }
             }
             for (auto& r : intersection_ranges) {
+                metas_added = true;
                 rlogger.debug("repair[{}] Repair tablet task table={}.{} master_shard_id={} range={} neighbors={} replicas={}",
                         rid.uuid(), keyspace_name, table_name, master_shard_id, r, repair_neighbors(nodes, shards).shard_map, m.replicas);
-                task_metas.push_back(tablet_repair_task_meta{keyspace_name, table_name, tid, master_shard_id, r, repair_neighbors(nodes, shards), m.replicas, erm});
-                co_await coroutine::maybe_yield();
+                co_await task_metas_builder.add_on_shard(master_shard_id, tablet_repair_task_meta{keyspace_name, table_name, tid, master_shard_id, r, repair_neighbors(nodes, shards), m.replicas});
             }
         }
+        if (metas_added) {
+            erms.push_back(std::move(erm));
+        }
     }
-    auto task = co_await _repair_module->make_and_start_task<repair::tablet_repair_task_impl>({}, rid, keyspace_name, tasks::task_id::create_null_id(), table_names, streaming::stream_reason::repair, std::move(task_metas), ranges_parallelism, service::default_session_id);
+    auto task = co_await _repair_module->make_and_start_task<repair::tablet_repair_task_impl>({}, rid, keyspace_name, tasks::task_id::create_null_id(), table_names, streaming::stream_reason::repair, co_await std::move(task_metas_builder).build(), ranges_parallelism, service::default_session_id, std::move(erms));
 }
 
 // It is called by the repair_tablet rpc verb to repair the given tablet
@@ -2494,11 +2579,11 @@ future<gc_clock::time_point> repair_service::repair_tablet(gms::gossip_address_m
         co_return flush_time;
     }
 
-    std::vector<tablet_repair_task_meta> task_metas;
+    repair::remote_metas_builder task_metas_builder;
     auto ranges_parallelism = std::nullopt;
     auto start = std::chrono::steady_clock::now();
-    task_metas.push_back(tablet_repair_task_meta{keyspace_name, table_name, table_id, *master_shard_id, range, repair_neighbors(nodes, shards), replicas});
-    auto task_impl_ptr = seastar::make_shared<repair::tablet_repair_task_impl>(_repair_module, id, keyspace_name, global_tablet_repair_task_info.id, table_names, streaming::stream_reason::repair, std::move(task_metas), ranges_parallelism, topo_guard, rebuild_replicas.has_value());
+    co_await task_metas_builder.add_on_shard(*master_shard_id, tablet_repair_task_meta{keyspace_name, table_name, table_id, *master_shard_id, range, repair_neighbors(nodes, shards), replicas});
+    auto task_impl_ptr = seastar::make_shared<repair::tablet_repair_task_impl>(_repair_module, id, keyspace_name, global_tablet_repair_task_info.id, table_names, streaming::stream_reason::repair, co_await std::move(task_metas_builder).build(), ranges_parallelism, topo_guard, utils::chunked_vector<locator::effective_replication_map_ptr>{}, rebuild_replicas.has_value());
     task_impl_ptr->sched_by_scheduler = true;
     auto task = co_await _repair_module->make_task(task_impl_ptr, global_tablet_repair_task_info);
     task->start();
@@ -2521,13 +2606,14 @@ tasks::is_user_task repair::tablet_repair_task_impl::is_user_task() const noexce
 
 future<> repair::tablet_repair_task_impl::release_resources() noexcept {
     _metas_size = _metas.size();
-    _metas = {};
+    _metas.clear();
     _tables = {};
+    _erms = {};
     return make_ready_future();
 }
 
 size_t repair::tablet_repair_task_impl::get_metas_size() const noexcept {
-    return _metas.size() > 0 ? _metas.size() : _metas_size;
+    return _metas_size == 0 ? _metas.size() : _metas_size;
 }
 
 future<> repair::tablet_repair_task_impl::run() {
@@ -2545,11 +2631,9 @@ future<> repair::tablet_repair_task_impl::run() {
         // Start the off strategy updater
         std::unordered_set<locator::host_id> participants;
         std::unordered_set<table_id> table_ids;
-        for (auto& meta : _metas) {
-            thread::maybe_yield();
-            participants.insert(meta.neighbors.all.begin(), meta.neighbors.all.end());
-            table_ids.insert(meta.tid);
-        }
+        auto hosts_and_tables = _metas.get_hosts_and_tables().get();
+        participants = std::move(hosts_and_tables.hosts);
+        table_ids = std::move(hosts_and_tables.tables);
         abort_source as;
         auto off_strategy_updater = seastar::async([&rs, uuid = id.uuid().uuid(), &table_ids, &participants, &as] {
             auto tables = std::list<table_id>(table_ids.begin(), table_ids.end());
@@ -2590,19 +2674,17 @@ future<> repair::tablet_repair_task_impl::run() {
 
         auto parent_shard = this_shard_id();
         std::vector<gc_clock::time_point> flush_times(smp::count);
-        rs.container().invoke_on_all([&idx, &flush_times, id, metas = _metas, parent_data, reason = _reason, tables = _tables, sched_by_scheduler = sched_by_scheduler, ranges_parallelism = _ranges_parallelism, parent_shard, topo_guard = _topo_guard, skip_flush = _skip_flush] (repair_service& rs) -> future<> {
+        rs.container().invoke_on_all([&idx, &flush_times, id, &metas = _metas, &parent_data, reason = _reason, &tables = _tables, sched_by_scheduler = sched_by_scheduler, ranges_parallelism = _ranges_parallelism, parent_shard, topo_guard = _topo_guard, skip_flush = _skip_flush] (repair_service& rs) -> future<> {
             std::exception_ptr error;
-            for (auto& m : metas) {
-                if (m.master_shard_id != this_shard_id()) {
-                    continue;
-                }
+            co_await metas.for_each_local_meta(coroutine::lambda([&rs, metas_size = metas.size(), &idx, id, &flush_times, parent_data, reason, &tables, sched_by_scheduler = sched_by_scheduler, ranges_parallelism, parent_shard, topo_guard, skip_flush, &error] (const tablet_repair_task_meta& m) -> future<> {
+                co_await coroutine::maybe_yield();
                 auto nr = idx.fetch_add(1);
                 rlogger.info("repair[{}] Repair {} out of {} tablets: table={}.{} range={} replicas={}",
-                    id.uuid(), nr, metas.size(), m.keyspace_name, m.table_name, m.range, m.replicas);
+                    id.uuid(), nr, metas_size, m.keyspace_name, m.table_name, m.range, m.replicas);
                 lw_shared_ptr<replica::table> t = rs._db.local().get_tables_metadata().get_table_if_exists(m.tid);
                 if (!t) {
                     rlogger.debug("repair[{}] Table {}.{} does not exist anymore", id.uuid(), m.keyspace_name, m.table_name);
-                    continue;
+                    co_return;
                 }
                 if (co_await rs.get_repair_module().is_aborted(id.uuid(), parent_shard)) {
                     throw abort_requested_exception();
@@ -2650,7 +2732,7 @@ future<> repair::tablet_repair_task_impl::run() {
                 auto current = flush_times[this_shard_id()];
                 auto time = task_impl_ptr->get_flush_time();
                 flush_times[this_shard_id()] = current == gc_clock::time_point() ? time : std::min(current, time);
-            }
+            }));
             if (error) {
                 co_await coroutine::return_exception_ptr(std::move(error));
             }
@@ -2677,10 +2759,6 @@ future<std::optional<double>> repair::tablet_repair_task_impl::expected_total_wo
     co_return sz ? std::make_optional<double>(sz) : std::nullopt;
 }
 
-std::optional<double> repair::tablet_repair_task_impl::expected_children_number() const {
-    return get_metas_size();
-}
-
 node_ops_cmd_category categorize_node_ops_cmd(node_ops_cmd cmd) noexcept {
     switch (cmd) {
     case node_ops_cmd::removenode_prepare:

repair/repair.hh

@@ -268,7 +268,6 @@ struct tablet_repair_task_meta {
     dht::token_range range;
     repair_neighbors neighbors;
     locator::tablet_replica_set replicas;
-    locator::effective_replication_map_ptr erm;
 };
 
 namespace std {

repair/row_level.cc

@@ -1448,7 +1448,9 @@ private:
         size_t row_bytes = co_await get_repair_rows_size(row_list);
         _metrics.tx_row_nr += row_list.size();
         _metrics.tx_row_bytes += row_bytes;
-        for (repair_row& r : row_list) {
+        while (!row_list.empty()) {
+            repair_row r = std::move(row_list.front());
+            row_list.pop_front();
             const auto& dk_with_hash = r.get_dk_with_hash();
             // No need to search from the beginning of the rows. Look at the end of repair_rows_on_wire is enough.
             if (rows.empty()) {
@@ -2762,7 +2764,12 @@ private:
 
     size_t get_max_row_buf_size(row_level_diff_detect_algorithm algo) {
         // Max buffer size per repair round
-        return is_rpc_stream_supported(algo) ?  repair::task_manager_module::max_repair_memory_per_range : 256 * 1024;
+        size_t size = is_rpc_stream_supported(algo) ? repair::task_manager_module::max_repair_memory_per_range : 256 * 1024;
+        if (_small_table_optimization) {
+            // For small table optimization, we reduce the buffer size to reduce memory consumption.
+            size /= _all_live_peer_nodes.size();
+        }
+        return size;
     }
 
     // Step A: Negotiate sync boundary to use
@@ -3096,7 +3103,7 @@ public:
 
             auto& mem_sem = _shard_task.rs.memory_sem();
             auto max = _shard_task.rs.max_repair_memory();
-            auto wanted = (_all_live_peer_nodes.size() + 1) * repair::task_manager_module::max_repair_memory_per_range;
+            auto wanted = (_all_live_peer_nodes.size() + 1) * max_row_buf_size;
             wanted = std::min(max, wanted);
             rlogger.trace("repair[{}]: Started to get memory budget, wanted={}, available={}, max_repair_memory={}",
                     _shard_task.global_repair_id.uuid(), wanted, mem_sem.current(), max);

repair/task_manager_module.hh

@@ -13,9 +13,12 @@
 #include "service/topology_guard.hh"
 #include "streaming/stream_reason.hh"
 #include "tasks/task_manager.hh"
+#include <cstddef>
 
 namespace repair {
 
+class remote_metas;
+
 class repair_task_impl : public tasks::task_manager::task::impl {
 protected:
     streaming::stream_reason _reason;
@@ -74,7 +77,6 @@ protected:
     future<> run() override;
 
     virtual future<std::optional<double>> expected_total_workload() const override;
-    virtual std::optional<double> expected_children_number() const override;
 };
 
 class data_sync_repair_task_impl : public repair_task_impl {
@@ -103,30 +105,72 @@ protected:
     future<> run() override;
 
     virtual future<std::optional<double>> expected_total_workload() const override;
-    virtual std::optional<double> expected_children_number() const override;
+};
+
+struct hosts_and_tables {
+    std::unordered_set<locator::host_id> hosts;
+    std::unordered_set<table_id> tables;
+};
+
+struct remote_data {
+    utils::chunked_vector<tablet_repair_task_meta> metas;
+};
+
+class remote_metas {
+public:
+    using remote_data_ptr = foreign_ptr<lw_shared_ptr<remote_data>>;
+private:
+    std::vector<remote_data_ptr> _metas_on_shards;
+    size_t _metas_count = 0;
+
+    remote_metas() : _metas_on_shards(smp::count) {}
+public:
+    future<> for_each_local_meta(std::function<future<>(const tablet_repair_task_meta&)> func) const;
+    size_t size() const;
+    void clear();
+    future<hosts_and_tables> get_hosts_and_tables() const;
+
+    friend class remote_metas_builder;
+};
+
+class remote_metas_builder {
+private:
+    remote_metas _remote_metas;
+    std::unordered_map<size_t, std::vector<tablet_repair_task_meta>> _pending_metas;
+    constexpr static size_t max_pending_metas_per_shard = 32;
+public:
+    remote_metas_builder() : _remote_metas() {}
+
+    future<> add_on_shard(size_t shard_id, tablet_repair_task_meta meta);
+    future<remote_metas> build() &&;
+private:
+    future<> flush(size_t shard_id);
+    future<> allocate_on_shard(size_t shard_id);
 };
 
 class tablet_repair_task_impl : public repair_task_impl {
 private:
     sstring _keyspace;
     std::vector<sstring> _tables;
-    std::vector<tablet_repair_task_meta> _metas;
+    remote_metas _metas;
     optimized_optional<abort_source::subscription> _abort_subscription;
     std::optional<int> _ranges_parallelism;
     size_t _metas_size = 0;
     gc_clock::time_point _flush_time;
     service::frozen_topology_guard _topo_guard;
+    utils::chunked_vector<locator::effective_replication_map_ptr> _erms;
     bool _skip_flush;
 public:
     bool sched_by_scheduler = false;
 public:
-    tablet_repair_task_impl(tasks::task_manager::module_ptr module, repair_uniq_id id, sstring keyspace, tasks::task_id parent_id, std::vector<sstring> tables, streaming::stream_reason reason, std::vector<tablet_repair_task_meta> metas, std::optional<int> ranges_parallelism, service::frozen_topology_guard topo_guard, bool skip_flush = false)
+    tablet_repair_task_impl(tasks::task_manager::module_ptr module, repair_uniq_id id, sstring keyspace, tasks::task_id parent_id, std::vector<sstring> tables, streaming::stream_reason reason, remote_metas metas, std::optional<int> ranges_parallelism, service::frozen_topology_guard topo_guard, utils::chunked_vector<locator::effective_replication_map_ptr> erms, bool skip_flush = false)
         : repair_task_impl(module, id.uuid(), id.id, "keyspace", keyspace, "", "", parent_id, reason)
         , _keyspace(std::move(keyspace))
         , _tables(std::move(tables))
         , _metas(std::move(metas))
         , _ranges_parallelism(ranges_parallelism)
         , _topo_guard(topo_guard)
+        , _erms(std::move(erms))
         , _skip_flush(skip_flush)
     {
     }
@@ -145,7 +189,6 @@ protected:
     future<> run() override;
 
     virtual future<std::optional<double>> expected_total_workload() const override;
-    virtual std::optional<double> expected_children_number() const override;
 };
 
 class shard_repair_task_impl : public repair_task_impl {

replica/database.cc

@@ -355,7 +355,7 @@ database::view_update_read_concurrency_sem() {
     return *sem;
 }
 
-database::database(const db::config& cfg, database_config dbcfg, service::migration_notifier& mn, gms::feature_service& feat, const locator::shared_token_metadata& stm,
+database::database(const db::config& cfg, database_config dbcfg, service::migration_notifier& mn, gms::feature_service& feat, locator::shared_token_metadata& stm,
         compaction_manager& cm, sstables::storage_manager& sstm, lang::manager& langm, sstables::directory_semaphore& sst_dir_sem, sstable_compressor_factory& scf, const abort_source& abort, utils::cross_shard_barrier barrier)
     : _stats(make_lw_shared<db_stats>())
     , _user_types(std::make_shared<db_user_types_storage>(*this))
@@ -2671,7 +2671,8 @@ future<> database::truncate(db::system_keyspace& sys_ks, column_family& cf, cons
     // since we don't want to leave behind data on disk with RP lower than the one we set
     // in the truncation table.
     if (st.did_flush && rp != db::replay_position() && st.low_mark < rp) {
-        on_internal_error(dblog, "Data written after truncation time was incorrectly truncated. Truncate is known to not work well with concurrent writes. Retry!");
+        dblog.warn("Data in table {}.{} is written after truncation time and was incorrectly truncated. truncated_at: {} low_mark: {} rp: {}",
+                    cf.schema()->ks_name(), cf.schema()->cf_name(), truncated_at, st.low_mark, rp);
     }
     if (rp == db::replay_position()) {
         // If this shard had no mutations, st.low_mark will be an empty, default constructed

replica/database.hh

@@ -1599,7 +1599,7 @@ private:
     service::migration_notifier& _mnotifier;
     gms::feature_service& _feat;
     std::vector<std::any> _listeners;
-    const locator::shared_token_metadata& _shared_token_metadata;
+    locator::shared_token_metadata& _shared_token_metadata;
     lang::manager& _lang_manager;
 
     reader_concurrency_semaphore_group _reader_concurrency_semaphores_group;
@@ -1684,7 +1684,7 @@ public:
     // (keyspace/table definitions, column mappings etc.)
     future<> parse_system_tables(distributed<service::storage_proxy>&, sharded<db::system_keyspace>&);
 
-    database(const db::config&, database_config dbcfg, service::migration_notifier& mn, gms::feature_service& feat, const locator::shared_token_metadata& stm,
+    database(const db::config&, database_config dbcfg, service::migration_notifier& mn, gms::feature_service& feat, locator::shared_token_metadata& stm,
             compaction_manager& cm, sstables::storage_manager& sstm, lang::manager& langm, sstables::directory_semaphore& sst_dir_sem, sstable_compressor_factory&, const abort_source& abort,
             utils::cross_shard_barrier barrier = utils::cross_shard_barrier(utils::cross_shard_barrier::solo{}) /* for single-shard usage */);
     database(database&&) = delete;
@@ -1719,7 +1719,7 @@ public:
         return _compaction_manager;
     }
 
-    const locator::shared_token_metadata& get_shared_token_metadata() const { return _shared_token_metadata; }
+    locator::shared_token_metadata& get_shared_token_metadata() const { return _shared_token_metadata; }
     locator::token_metadata_ptr get_token_metadata_ptr() const { return _shared_token_metadata.get(); }
     const locator::token_metadata& get_token_metadata() const { return *_shared_token_metadata.get(); }
 

replica/table.cc

@@ -2737,7 +2737,17 @@ void tablet_storage_group_manager::update_effective_replication_map(const locato
             _storage_groups[tid.value()] = allocate_storage_group(*new_tablet_map, tid, std::move(range));
             tablet_migrating_in = true;
         } else if (_storage_groups.contains(tid.value()) && locator::is_post_cleanup(this_replica, new_tablet_map->get_tablet_info(tid), transition_info)) {
+            // The storage group should be cleaned up and stopped at this point usually by the tablet cleanup stage,
+            // unless the storage group was allocated after tablet cleanup was completed for this node. This could
+            // happen if the node was restarted after tablet cleanup was run but before moving to the next stage. To
+            // handle this case we stop the storage group here if it's not stopped already.
+            auto sg = _storage_groups[tid.value()];
+
             remove_storage_group(tid.value());
+
+            (void) with_gate(_t.async_gate(), [sg] {
+                return sg->stop("tablet post-cleanup").then([sg] {});
+            });
         }
     }
 

scylla-gdb.py

@@ -999,6 +999,8 @@ class managed_bytes:
         inf = gdb.selected_inferior()
 
         def to_bytes(data, size):
+            if size == 0:
+                return b''
             return bytes(inf.read_memory(data, size))
 
         if self.is_inline():

service/migration_manager.cc

@@ -56,7 +56,9 @@ migration_manager::migration_manager(migration_notifier& notifier, gms::feature_
         , _group0_barrier(this_shard_id() == 0 ?
             std::function<future<>()>([this] () -> future<> {
                 // This will run raft barrier and will sync schema with the leader
-                (void)co_await start_group0_operation();
+                return with_scheduling_group(_storage_proxy.get_db().local().get_gossip_scheduling_group(), [this] {
+                    return start_group0_operation().discard_result();
+                });
             }) :
             std::function<future<>()>([this] () -> future<> {
                 co_await container().invoke_on(0, [] (migration_manager& mm) -> future<> {

service/qos/raft_service_level_distributed_data_accessor.cc

@@ -100,6 +100,10 @@ bool raft_service_level_distributed_data_accessor::is_v2() const {
     return true;
 }
 
+bool raft_service_level_distributed_data_accessor::can_use_effective_service_level_cache() const {
+    return !auth::legacy_mode(_qp);
+}
+
 ::shared_ptr<service_level_controller::service_level_distributed_data_accessor> raft_service_level_distributed_data_accessor::upgrade_to_v2(cql3::query_processor& qp, service::raft_group0_client& group0_client) const {
     return nullptr;
 }

service/qos/raft_service_level_distributed_data_accessor.hh

@@ -39,6 +39,7 @@ public:
     virtual future<> commit_mutations(service::group0_batch&& mc, abort_source& as) const override;
 
     virtual bool is_v2() const override;
+    virtual bool can_use_effective_service_level_cache() const override;
     virtual ::shared_ptr<service_level_distributed_data_accessor> upgrade_to_v2(cql3::query_processor& qp, service::raft_group0_client& group0_client) const override;
 };
 

service/qos/service_level_controller.cc

@@ -304,8 +304,18 @@ future<> service_level_controller::update_effective_service_levels_cache() {
     SCYLLA_ASSERT(this_shard_id() == global_controller);
     
     if (!_auth_service.local_is_initialized()) {
-        // Because cache update is triggered in `topology_state_load()`, auth service
-        // might be not initialized yet.
+        // Auth service might be not initialized yet.
+        co_return;
+    }
+    if (!_sl_data_accessor || !_sl_data_accessor->can_use_effective_service_level_cache()) {
+        // Don't populate the effective service level cache until auth is migrated to raft.
+        // Otherwise, executing the code that follows would read roles data
+        // from system_auth tables; that would be bad because reading from
+        // those tables is prone to timeouts, and `update_effective_service_levels_cache`
+        // is called from the group0 context - a timeout like that would render
+        // group0 non-functional on the node until restart.
+        //
+        // See scylladb/scylladb#24963 for more details.
         co_return;
     }
     auto units = co_await get_units(_global_controller_db->notifications_serializer, 1);
@@ -384,7 +394,7 @@ void service_level_controller::stop_legacy_update_from_distributed_data() {
 }
 
 future<std::optional<service_level_options>> service_level_controller::find_effective_service_level(const sstring& role_name) {
-    if (_sl_data_accessor->is_v2()) {
+    if (_sl_data_accessor->can_use_effective_service_level_cache()) {
         auto effective_sl_it = _effective_service_levels_db.find(role_name);
         co_return effective_sl_it != _effective_service_levels_db.end() 
             ? std::optional<service_level_options>(effective_sl_it->second)

service/qos/service_level_controller.hh

@@ -114,6 +114,9 @@ public:
         virtual future<> commit_mutations(service::group0_batch&& mc, abort_source& as) const = 0;
 
         virtual bool is_v2() const = 0;
+        // Returns whether effective service level cache can be populated and used.
+        // This is equivalent to checking whether auth + raft have been migrated to raft.
+        virtual bool can_use_effective_service_level_cache() const = 0;
         // Returns v2(raft) data accessor. If data accessor is already a raft one, returns nullptr.
         virtual ::shared_ptr<service_level_distributed_data_accessor> upgrade_to_v2(cql3::query_processor& qp, service::raft_group0_client& group0_client) const = 0;
     };

service/qos/standard_service_level_distributed_data_accessor.cc

@@ -37,6 +37,10 @@ bool standard_service_level_distributed_data_accessor::is_v2() const {
     return false;
 }
 
+bool standard_service_level_distributed_data_accessor::can_use_effective_service_level_cache() const {
+    return false;
+}
+
 ::shared_ptr<service_level_controller::service_level_distributed_data_accessor> standard_service_level_distributed_data_accessor::upgrade_to_v2(cql3::query_processor& qp, service::raft_group0_client& group0_client) const {
     return ::static_pointer_cast<service_level_controller::service_level_distributed_data_accessor>(
                 ::make_shared<raft_service_level_distributed_data_accessor>(qp, group0_client));

service/qos/standard_service_level_distributed_data_accessor.hh

@@ -31,6 +31,7 @@ public:
     virtual future<> commit_mutations(service::group0_batch&& mc, abort_source& as) const override { return make_ready_future(); }
 
     virtual bool is_v2() const override;
+    virtual bool can_use_effective_service_level_cache() const override;
     virtual ::shared_ptr<service_level_distributed_data_accessor> upgrade_to_v2(cql3::query_processor& qp, service::raft_group0_client& group0_client) const override;
 };
 }

service/raft/group0_state_machine.cc

@@ -333,6 +333,7 @@ future<> group0_state_machine::load_snapshot(raft::snapshot_id id) {
     if (_feature_service.compression_dicts) {
         co_await _ss.compression_dictionary_updated_callback_all();
     }
+    co_await _ss.update_service_levels_cache(qos::update_both_cache_levels::yes, qos::query_context::group0);
     _ss._topology_state_machine.event.broadcast();
 }
 

service/raft/raft_group0.cc

@@ -414,12 +414,6 @@ future<group0_info> persistent_discovery::run(
 }
 
 future<> raft_group0::abort() {
-    if (_aborted) {
-        co_return;
-    }
-    _aborted = true;
-    group0_log.debug("Raft group0 service is aborting...");
-
     co_await smp::invoke_on_all([this]() {
         return uninit_rpc_verbs(_ms.local());
     });
@@ -431,8 +425,6 @@ future<> raft_group0::abort() {
     co_await std::move(_leadership_monitor);
 
     co_await stop_group0();
-
-    group0_log.debug("Raft group0 service is aborted");
 }
 
 future<> raft_group0::start_server_for_group0(raft::group_id group0_id, service::storage_service& ss, cql3::query_processor& qp, service::migration_manager& mm, bool topology_change_enabled) {
@@ -543,7 +535,7 @@ future<> raft_group0::join_group0(std::vector<gms::inet_address> seeds, shared_p
                 // created in the Raft-based recovery procedure). The persistent topology state is present on that node
                 // when it creates the new group 0. Also, it joins the new group 0 using legacy_handshaker, so there is
                 // no need to create a join request.
-                if (topology_change_enabled && qp.db().get_config().recovery_leader().empty()) {
+                if (topology_change_enabled && !qp.db().get_config().recovery_leader.is_set()) {
                     co_await ss.raft_initialize_discovery_leader(params);
                 }
 
@@ -718,7 +710,7 @@ future<> raft_group0::setup_group0_if_exist(db::system_keyspace& sys_ks, service
         } else {
             // We'll disable them once we complete the upgrade procedure.
         }
-    } else if (qp.db().get_config().recovery_leader().empty()) {
+    } else if (!qp.db().get_config().recovery_leader.is_set()) {
         // Scylla has bootstrapped earlier but group 0 ID is not present and we are not recovering from majority loss
         // using the Raft-based procedure. This means we're upgrading.
         // Upgrade will start through a feature listener created after we enter NORMAL state.

service/raft/raft_group0.hh

@@ -133,7 +133,6 @@ class raft_group0 {
     future<> _leadership_monitor = make_ready_future<>();
     abort_source _leadership_monitor_as;
     utils::updateable_value_source<bool> _leadership_observable;
-    bool _aborted = false;
 
 public:
     // Passed to `setup_group0` when replacing a node.

service/storage_proxy.cc

@@ -600,6 +600,8 @@ private:
         ++p->get_stats().received_mutations;
         p->get_stats().forwarded_mutations += forward_host_id.size();
 
+        co_await utils::get_local_injector().inject("storage_proxy_write_response_pause", utils::wait_for_message(5min));
+
         if (auto stale = _sp.apply_fence(fence, src_addr)) {
             errors.count += (forward_host_id.size() + 1);
             errors.local = std::move(*stale);
@@ -1101,26 +1103,23 @@ private:
 
             global_request_id = guard.new_group0_state_id();
 
-            std::vector<canonical_mutation> updates;
             topology_mutation_builder builder(guard.write_timestamp());
+            topology_request_tracking_mutation_builder trbuilder(global_request_id, _sp._features.topology_requests_type_column);
+            trbuilder.set_truncate_table_data(table_id)
+                     .set("done", false)
+                     .set("start_time", db_clock::now());
+
             if (!_sp._features.topology_global_request_queue) {
                 builder.set_global_topology_request(global_topology_request::truncate_table)
                        .set_global_topology_request_id(global_request_id);
             } else {
                 builder.queue_global_topology_request_id(global_request_id);
+                trbuilder.set("request_type", global_topology_request::truncate_table);
             }
-            updates.emplace_back(builder.build());
-
-            updates.emplace_back(topology_request_tracking_mutation_builder(global_request_id, _sp._features.topology_requests_type_column)
-                                    .set_truncate_table_data(table_id)
-                                    .set("done", false)
-                                    .set("start_time", db_clock::now())
-                                    .set("request_type", global_topology_request::truncate_table)
-                                    .build());
 
             slogger.info("Creating TRUNCATE global topology request for table {}.{}", ks_name, cf_name);
 
-            topology_change change{std::move(updates)};
+            topology_change change{{builder.build(), trbuilder.build()}};
             sstring reason = "Truncating table";
             group0_command g0_cmd = _group0_client.prepare_command(std::move(change), guard, reason);
             try {
@@ -1615,6 +1614,10 @@ public:
         return _type == db::write_type::VIEW;
     }
 
+    bool is_batch() const noexcept {
+        return _type == db::write_type::BATCH;
+    }
+
     void set_cdc_operation_result_tracker(lw_shared_ptr<cdc::operation_result_tracker> tracker) {
         _cdc_operation_result_tracker = std::move(tracker);
     }
@@ -2120,7 +2123,7 @@ paxos_response_handler::begin_and_repair_paxos(client_state& cs, unsigned& conte
             // create_write_response_handler is overloaded for paxos::proposal and will
             // create cas_mutation holder, which consequently will ensure paxos::learn is
             // used.
-            auto f = _proxy->mutate_internal(std::move(m), db::consistency_level::ANY, false, tr_state, _permit, _timeout)
+            auto f = _proxy->mutate_internal(std::move(m), db::consistency_level::ANY, tr_state, _permit, _timeout)
                     .then(utils::result_into_future<result<>>);
 
             // TODO: provided commits did not invalidate the prepare we just did above (which they
@@ -2472,7 +2475,7 @@ future<> paxos_response_handler::learn_decision(lw_shared_ptr<paxos::proposal> d
                 return v.schema()->id() == base_tbl_id;
             });
             if (!mutations.empty()) {
-                f_cdc = _proxy->mutate_internal(std::move(mutations), _cl_for_learn, false, tr_state, _permit, _timeout, std::move(tracker))
+                f_cdc = _proxy->mutate_internal(std::move(mutations), _cl_for_learn, tr_state, _permit, _timeout, {}, std::move(tracker))
                         .then(utils::result_into_future<result<>>);
             }
         }
@@ -2480,7 +2483,7 @@ future<> paxos_response_handler::learn_decision(lw_shared_ptr<paxos::proposal> d
 
     // Path for the "base" mutations
     std::array<std::tuple<lw_shared_ptr<paxos::proposal>, schema_ptr, shared_ptr<paxos_response_handler>, dht::token>, 1> m{std::make_tuple(std::move(decision), _schema, shared_from_this(), _key.token())};
-    future<> f_lwt = _proxy->mutate_internal(std::move(m), _cl_for_learn, false, tr_state, _permit, _timeout)
+    future<> f_lwt = _proxy->mutate_internal(std::move(m), _cl_for_learn, tr_state, _permit, _timeout)
             .then(utils::result_into_future<result<>>);
 
     co_await when_all_succeed(std::move(f_cdc), std::move(f_lwt)).discard_result();
@@ -3071,6 +3074,10 @@ struct hint_wrapper {
     mutation mut;
 };
 
+struct batchlog_replay_mutation {
+    mutation mut;
+};
+
 struct read_repair_mutation {
     std::unordered_map<locator::host_id, std::optional<mutation>> value;
     locator::effective_replication_map_ptr ermp;
@@ -3084,6 +3091,12 @@ template <> struct fmt::formatter<service::hint_wrapper> : fmt::formatter<string
     }
 };
 
+template <> struct fmt::formatter<service::batchlog_replay_mutation> : fmt::formatter<string_view> {
+    auto format(const service::batchlog_replay_mutation& h, fmt::format_context& ctx) const {
+        return fmt::format_to(ctx.out(), "batchlog_replay_mutation{{{}}}", h.mut);
+    }
+};
+
 template <>
 struct fmt::formatter<service::read_repair_mutation> : fmt::formatter<string_view> {
     auto format(const service::read_repair_mutation& m, fmt::format_context& ctx) const {
@@ -3449,6 +3462,12 @@ storage_proxy::create_write_response_handler(const hint_wrapper& h, db::consiste
             std::move(permit), allow_limit, is_cancellable::yes);
 }
 
+result<storage_proxy::response_id_type>
+storage_proxy::create_write_response_handler(const batchlog_replay_mutation& m, db::consistency_level cl, db::write_type type, tracing::trace_state_ptr tr_state, service_permit permit, db::allow_per_partition_rate_limit allow_limit) {
+    return create_write_response_handler_helper(m.mut.schema(), m.mut.token(), std::make_unique<shared_mutation>(m.mut), cl, type, tr_state,
+            std::move(permit), allow_limit, is_cancellable::yes);
+}
+
 result<storage_proxy::response_id_type>
 storage_proxy::create_write_response_handler(const read_repair_mutation& mut, db::consistency_level cl, db::write_type type, tracing::trace_state_ptr tr_state, service_permit permit, db::allow_per_partition_rate_limit allow_limit) {
     host_id_vector_replica_set endpoints;
@@ -3843,7 +3862,7 @@ future<result<>> storage_proxy::do_mutate(std::vector<mutation> mutations, db::c
     }).begin();
     return seastar::when_all_succeed(
         mutate_counters(std::ranges::subrange(mutations.begin(), mid), cl, tr_state, permit, timeout),
-        mutate_internal(std::ranges::subrange(mid, mutations.end()), cl, false, tr_state, permit, timeout, std::move(cdc_tracker), allow_limit)
+        mutate_internal(std::ranges::subrange(mid, mutations.end()), cl, tr_state, permit, timeout, {}, std::move(cdc_tracker), allow_limit)
     ).then([] (std::tuple<result<>> res) {
         // For now, only mutate_internal returns a result<>
         return std::get<0>(std::move(res));
@@ -3852,8 +3871,10 @@ future<result<>> storage_proxy::do_mutate(std::vector<mutation> mutations, db::c
 
 future<> storage_proxy::replicate_counter_from_leader(mutation m, db::consistency_level cl, tracing::trace_state_ptr tr_state,
                                                       clock_type::time_point timeout, service_permit permit) {
+    // we need to pass correct db::write_type in case of a timeout so that
+    // client doesn't attempt to retry the request.
     // FIXME: do not send the mutation to itself, it has already been applied (it is not incorrect to do so, though)
-    return mutate_internal(std::array<mutation, 1>{std::move(m)}, cl, true, std::move(tr_state), std::move(permit), timeout)
+    return mutate_internal(std::array<mutation, 1>{std::move(m)}, cl, std::move(tr_state), std::move(permit), timeout, db::write_type::COUNTER)
             .then(utils::result_into_future<result<>>);
 }
 
@@ -3864,8 +3885,8 @@ future<> storage_proxy::replicate_counter_from_leader(mutation m, db::consistenc
  */
 template<typename Range>
 future<result<>>
-storage_proxy::mutate_internal(Range mutations, db::consistency_level cl, bool counters, tracing::trace_state_ptr tr_state, service_permit permit,
-                               std::optional<clock_type::time_point> timeout_opt, lw_shared_ptr<cdc::operation_result_tracker> cdc_tracker,
+storage_proxy::mutate_internal(Range mutations, db::consistency_level cl, tracing::trace_state_ptr tr_state, service_permit permit,
+                               std::optional<clock_type::time_point> timeout_opt, std::optional<db::write_type> type_opt, lw_shared_ptr<cdc::operation_result_tracker> cdc_tracker,
                                db::allow_per_partition_rate_limit allow_limit) {
     if (std::ranges::empty(mutations)) {
         return make_ready_future<result<>>(bo::success());
@@ -3874,12 +3895,10 @@ storage_proxy::mutate_internal(Range mutations, db::consistency_level cl, bool c
     slogger.trace("mutate cl={}", cl);
     mlogger.trace("mutations={}", mutations);
 
-    // If counters is set it means that we are replicating counter shards. There
-    // is no need for special handling anymore, since the leader has already
-    // done its job, but we need to return correct db::write_type in case of
-    // a timeout so that client doesn't attempt to retry the request.
-    auto type = counters ? db::write_type::COUNTER
-                         : (std::next(std::begin(mutations)) == std::end(mutations) ? db::write_type::SIMPLE : db::write_type::UNLOGGED_BATCH);
+    // the parameter type_opt allows to pass a specific type if needed for
+    // special handling, e.g. counters. otherwise, a default type is used.
+    auto type = type_opt.value_or(std::next(std::begin(mutations)) == std::end(mutations) ? db::write_type::SIMPLE : db::write_type::UNLOGGED_BATCH);
+
     utils::latency_counter lc;
     lc.start();
 
@@ -4065,6 +4084,7 @@ storage_proxy::mutate_atomically_result(std::vector<mutation> mutations, db::con
         };
         future<> async_remove_from_batchlog() {
             // delete batch
+            utils::get_local_injector().inject("storage_proxy_fail_remove_from_batchlog", [] { throw std::runtime_error("Error injection: failing remove from batchlog"); });
             auto key = partition_key::from_exploded(*_schema, {uuid_type->decompose(_batch_uuid)});
             auto now = service::client_state(service::client_state::internal_tag()).get_timestamp();
             mutation m(_schema, key);
@@ -4136,13 +4156,15 @@ mutation storage_proxy::do_get_batchlog_mutation_for(schema_ptr schema, const st
         for (auto& m : fm) {
             ser::serialize(out, m);
         }
-        return to_bytes(out.linearize());
+        return std::move(out).to_managed_bytes();
     }();
 
     mutation m(schema, key);
     m.set_cell(clustering_key_prefix::make_empty(), to_bytes("version"), version, timestamp);
     m.set_cell(clustering_key_prefix::make_empty(), to_bytes("written_at"), now, timestamp);
-    m.set_cell(clustering_key_prefix::make_empty(), to_bytes("data"), data_value(std::move(data)), timestamp);
+    // Avoid going through data_value and therefore `bytes`, as it can be large (#24809).
+    auto cdef_data = schema->get_column_definition(to_bytes("data"));
+    m.set_cell(clustering_key_prefix::make_empty(), *cdef_data, atomic_cell::make_live(*cdef_data->type, timestamp, std::move(data)));
 
     return m;
 }
@@ -4248,7 +4270,16 @@ future<> storage_proxy::send_hint_to_endpoint(frozen_mutation_and_schema fm_a_s,
 
 future<> storage_proxy::send_hint_to_all_replicas(frozen_mutation_and_schema fm_a_s) {
     std::array<hint_wrapper, 1> ms{hint_wrapper { fm_a_s.fm.unfreeze(fm_a_s.s) }};
-    return mutate_internal(std::move(ms), db::consistency_level::ALL, false, nullptr, empty_service_permit())
+    return mutate_internal(std::move(ms), db::consistency_level::ALL, nullptr, empty_service_permit())
+            .then(utils::result_into_future<result<>>);
+}
+
+future<> storage_proxy::send_batchlog_replay_to_all_replicas(std::vector<mutation> mutations, clock_type::time_point timeout) {
+    std::vector<batchlog_replay_mutation> ms = mutations | std::views::transform([] (auto&& m) {
+            return batchlog_replay_mutation(std::move(m));
+        }) | std::ranges::to<std::vector<batchlog_replay_mutation>>();
+
+    return mutate_internal(std::move(ms), db::consistency_level::ALL, nullptr, empty_service_permit(), timeout, db::write_type::BATCH)
             .then(utils::result_into_future<result<>>);
 }
 
@@ -4431,7 +4462,7 @@ future<result<>> storage_proxy::schedule_repair(locator::effective_replication_m
                     std::views::transform([ermp] (auto& v) { return read_repair_mutation{std::move(v), ermp}; }) |
                     // The transform above is destructive, materialize into a vector to make the range re-iterable.
                     std::ranges::to<std::vector<read_repair_mutation>>()
-            , cl, false, std::move(trace_state), std::move(permit));
+            , cl, std::move(trace_state), std::move(permit));
 }
 
 class abstract_read_resolver {
@@ -6953,7 +6984,7 @@ future<> storage_proxy::drain_on_shutdown() {
     //NOTE: the thread is spawned here because there are delicate lifetime issues to consider
     // and writing them down with plain futures is error-prone.
     return async([this] {
-        cancel_write_handlers([] (const abstract_write_response_handler&) { return true; });
+        cancel_all_write_response_handlers().get();
         _hints_resource_manager.stop().get();
     });
 }
@@ -6964,6 +6995,12 @@ future<> storage_proxy::abort_view_writes() {
     });
 }
 
+future<> storage_proxy::abort_batch_writes() {
+    return async([this] {
+        cancel_write_handlers([] (const abstract_write_response_handler& handler) { return handler.is_batch(); });
+    });
+}
+
 future<>
 storage_proxy::stop() {
     return make_ready_future<>();
@@ -6977,4 +7014,13 @@ future<utils::chunked_vector<dht::token_range_endpoints>> storage_proxy::describ
     return locator::describe_ring(_db.local(), _remote->gossiper(), keyspace, include_only_local_dc);
 }
 
+future<> storage_proxy::cancel_all_write_response_handlers() {
+    while (!_response_handlers.empty()) {
+        _response_handlers.begin()->second->timeout_cb();
+
+        if (!_response_handlers.empty()) {
+            co_await maybe_yield();
+        }
+    }
+}
 }

service/storage_proxy.hh

@@ -87,6 +87,7 @@ class mutation_holder;
 class client_state;
 class migration_manager;
 struct hint_wrapper;
+struct batchlog_replay_mutation;
 struct read_repair_mutation;
 
 using replicas_per_token_range = std::unordered_map<dht::token_range, std::vector<locator::host_id>>;
@@ -340,6 +341,7 @@ private:
             const host_id_vector_topology_change& pending_endpoints, host_id_vector_topology_change, tracing::trace_state_ptr tr_state, storage_proxy::write_stats& stats, service_permit permit, db::per_partition_rate_limit::info rate_limit_info, is_cancellable);
     result<response_id_type> create_write_response_handler(const mutation&, db::consistency_level cl, db::write_type type, tracing::trace_state_ptr tr_state, service_permit permit, db::allow_per_partition_rate_limit allow_limit);
     result<response_id_type> create_write_response_handler(const hint_wrapper&, db::consistency_level cl, db::write_type type, tracing::trace_state_ptr tr_state, service_permit permit, db::allow_per_partition_rate_limit allow_limit);
+    result<response_id_type> create_write_response_handler(const batchlog_replay_mutation&, db::consistency_level cl, db::write_type type, tracing::trace_state_ptr tr_state, service_permit permit, db::allow_per_partition_rate_limit allow_limit);
     result<response_id_type> create_write_response_handler(const read_repair_mutation&, db::consistency_level cl, db::write_type type, tracing::trace_state_ptr tr_state, service_permit permit, db::allow_per_partition_rate_limit allow_limit);
     result<response_id_type> create_write_response_handler(const std::tuple<lw_shared_ptr<paxos::proposal>, schema_ptr, shared_ptr<paxos_response_handler>, dht::token>& proposal,
             db::consistency_level cl, db::write_type type, tracing::trace_state_ptr tr_state, service_permit permit, db::allow_per_partition_rate_limit allow_limit);
@@ -427,7 +429,7 @@ private:
     void unthrottle();
     void handle_read_error(std::variant<exceptions::coordinator_exception_container, std::exception_ptr> failure, bool range);
     template<typename Range>
-    future<result<>> mutate_internal(Range mutations, db::consistency_level cl, bool counter_write, tracing::trace_state_ptr tr_state, service_permit permit, std::optional<clock_type::time_point> timeout_opt = { }, lw_shared_ptr<cdc::operation_result_tracker> cdc_tracker = { }, db::allow_per_partition_rate_limit allow_limit = db::allow_per_partition_rate_limit::no);
+    future<result<>> mutate_internal(Range mutations, db::consistency_level cl, tracing::trace_state_ptr tr_state, service_permit permit, std::optional<clock_type::time_point> timeout_opt = { }, std::optional<db::write_type> type = { }, lw_shared_ptr<cdc::operation_result_tracker> cdc_tracker = { }, db::allow_per_partition_rate_limit allow_limit = db::allow_per_partition_rate_limit::no);
     future<rpc::tuple<foreign_ptr<lw_shared_ptr<reconcilable_result>>, cache_temperature>> query_nonsingular_mutations_locally(
             schema_ptr s, lw_shared_ptr<query::read_command> cmd, const dht::partition_range_vector&& pr, tracing::trace_state_ptr trace_state,
             clock_type::time_point timeout);
@@ -521,6 +523,8 @@ public:
     bool is_me(gms::inet_address addr) const noexcept;
     bool is_me(const locator::effective_replication_map& erm, locator::host_id id) const noexcept;
 
+    future<> cancel_all_write_response_handlers();
+
 private:
     bool only_me(const locator::effective_replication_map& erm, const host_id_vector_replica_set& replicas) const noexcept;
 
@@ -631,6 +635,8 @@ public:
 
     future<> send_hint_to_all_replicas(frozen_mutation_and_schema fm_a_s);
 
+    future<> send_batchlog_replay_to_all_replicas(std::vector<mutation> mutations, clock_type::time_point timeout);
+
     // Send a mutation to one specific remote target.
     // Inspired by Cassandra's StorageProxy.sendToHintedEndpoints but without
     // hinted handoff support, and just one target. See also
@@ -705,6 +711,7 @@ public:
     void allow_replaying_hints() noexcept;
     future<> drain_hints_for_left_nodes();
     future<> abort_view_writes();
+    future<> abort_batch_writes();
 
     future<> change_hints_host_filter(db::hints::host_filter new_filter);
     const db::hints::host_filter& get_hints_host_filter() const;

service/storage_service.cc

@@ -111,7 +111,6 @@
 #include "node_ops/task_manager_module.hh"
 #include "service/task_manager_module.hh"
 #include "service/topology_mutation.hh"
-#include "service/topology_coordinator.hh"
 #include "cql3/query_processor.hh"
 #include "service/qos/service_level_controller.hh"
 #include "service/qos/standard_service_level_distributed_data_accessor.hh"
@@ -707,7 +706,6 @@ future<> storage_service::topology_state_load(state_change_hint hint) {
     co_await _sl_controller.invoke_on_all([this] (qos::service_level_controller& sl_controller) {
         sl_controller.upgrade_to_v2(_qp, _group0->client());
     });
-    co_await update_service_levels_cache(qos::update_both_cache_levels::yes, qos::query_context::group0);
 
     // the view_builder is migrated to v2 in view_builder::migrate_to_v2.
     // it writes a v2 version mutation as topology_change, then we get here
@@ -740,9 +738,7 @@ future<> storage_service::topology_state_load(state_change_hint hint) {
     auto saved_tmpr = get_token_metadata_ptr();
     {
         auto tmlock = co_await get_token_metadata_lock();
-        auto tmptr = make_token_metadata_ptr(token_metadata::config {
-            get_token_metadata().get_topology().get_config()
-        });
+        auto tmptr = _shared_token_metadata.make_token_metadata_ptr();
         tmptr->invalidate_cached_rings();
 
         tmptr->set_version(_topology_state_machine._topology.version);
@@ -817,10 +813,6 @@ future<> storage_service::topology_state_load(state_change_hint hint) {
 
     for (const auto& gen_id : _topology_state_machine._topology.committed_cdc_generations) {
         rtlogger.trace("topology_state_load: process committed cdc generation {}", gen_id);
-        co_await utils::get_local_injector().inject("topology_state_load_before_update_cdc", [](auto& handler) -> future<> {
-            rtlogger.info("topology_state_load_before_update_cdc hit, wait for message");
-            co_await handler.wait_for_message(db::timeout_clock::now() + std::chrono::minutes(5));
-        });
         co_await _cdc_gens.local().handle_cdc_generation(gen_id);
         if (gen_id == _topology_state_machine._topology.committed_cdc_generations.back()) {
             co_await _sys_ks.local().update_cdc_generation_id(gen_id);
@@ -903,7 +895,10 @@ future<> storage_service::merge_topology_snapshot(raft_snapshot snp) {
 
 future<> storage_service::update_service_levels_cache(qos::update_both_cache_levels update_only_effective_cache, qos::query_context ctx) {
     SCYLLA_ASSERT(this_shard_id() == 0);
-    co_await _sl_controller.local().update_cache(update_only_effective_cache, ctx);
+    if (_sl_controller.local().is_v2()) {
+        // Skip cache update unless the topology upgrade is done
+        co_await _sl_controller.local().update_cache(update_only_effective_cache, ctx);
+    }
 }
 
 future<> storage_service::compression_dictionary_updated_callback_all() {
@@ -1134,7 +1129,8 @@ future<> storage_service::raft_state_monitor_fiber(raft::server& raft, gate::hol
                     _tablet_allocator.local(),
                     get_ring_delay(),
                     _lifecycle_notifier,
-                    _feature_service);
+                    _feature_service,
+                    _topology_cmd_rpc_tracker);
         }
     } catch (...) {
         rtlogger.info("raft_state_monitor_fiber aborted with {}", std::current_exception());
@@ -1773,7 +1769,7 @@ future<> storage_service::join_topology(sharded<service::storage_proxy>& proxy,
     // the topology coordinator. We can assume this node has already been accepted by the topology coordinator once
     // and joined topology.
     ::shared_ptr<group0_handshaker> handshaker =
-            raft_topology_change_enabled() && _db.local().get_config().recovery_leader().empty()
+            raft_topology_change_enabled() && !_db.local().get_config().recovery_leader.is_set()
             ? ::make_shared<join_node_rpc_handshaker>(*this, join_params)
             : _group0->make_legacy_handshaker(can_vote::no);
     co_await _group0->setup_group0(_sys_ks.local(), initial_contact_nodes, std::move(handshaker),
@@ -2936,7 +2932,7 @@ future<> storage_service::join_cluster(sharded<service::storage_proxy>& proxy,
 
     gms::inet_address recovery_leader_ip;
     locator::host_id recovery_leader_id;
-    if (!_db.local().get_config().recovery_leader().empty()) {
+    if (_db.local().get_config().recovery_leader.is_set()) {
         if (_group0->joined_group0()) {
             // Something is wrong unless it is a noninitial (and unneeded) restart while recreating the new group 0 in
             // the Raft-based recovery procedure.
@@ -2946,7 +2942,7 @@ future<> storage_service::join_cluster(sharded<service::storage_proxy>& proxy,
                     "the Raft-based recovery procedure, please follow the steps in the documentation.",
                     _db.local().get_config().recovery_leader(), _group0->load_my_id());
         } else {
-            recovery_leader_id = locator::host_id(utils::UUID(_db.local().get_config().recovery_leader()));
+            recovery_leader_id = locator::host_id(_db.local().get_config().recovery_leader());
             auto recovery_leader_it = loaded_endpoints.find(recovery_leader_id);
             if (recovery_leader_id != my_host_id() && recovery_leader_it == loaded_endpoints.end()) {
                 throw std::runtime_error(
@@ -3146,9 +3142,10 @@ future<> storage_service::replicate_to_all_cores(mutable_token_metadata_ptr tmpt
     try {
         auto base_shard = this_shard_id();
         pending_token_metadata_ptr[base_shard] = tmptr;
+        auto& sharded_token_metadata = _shared_token_metadata.container();
         // clone a local copy of updated token_metadata on all other shards
         co_await smp::invoke_on_others(base_shard, [&, tmptr] () -> future<> {
-            pending_token_metadata_ptr[this_shard_id()] = make_token_metadata_ptr(co_await tmptr->clone_async());
+            pending_token_metadata_ptr[this_shard_id()] = sharded_token_metadata.local().make_token_metadata_ptr(co_await tmptr->clone_async());
         });
 
         // Precalculate new effective_replication_map for all keyspaces
@@ -4701,17 +4698,13 @@ future<> storage_service::drain() {
 }
 
 future<> storage_service::do_drain() {
-    // Need to stop transport before group0, otherwise RPCs may fail with raft_group_not_found.
     co_await stop_transport();
 
-    // group0 persistence relies on local storage, so we need to stop group0 first.
-    // This must be kept in sync with defer_verbose_shutdown for group0 in main.cc to
-    // handle the case when initialization fails before reaching drain_on_shutdown for ss.
-    _sl_controller.local().abort_group0_operations();
+    // Drain view builder before group0, because the view builder uses group0 to coordinate view building.
+    // Drain after transport is stopped, because view_builder::drain aborts view writes for user writes as well.
+    co_await _view_builder.invoke_on_all(&db::view::view_builder::drain);
+
     co_await wait_for_group0_stop();
-    if (_group0) {
-        co_await _group0->abort();
-    }
 
     co_await tracing::tracing::tracing_instance().invoke_on_all(&tracing::tracing::shutdown);
 
@@ -4719,7 +4712,6 @@ future<> storage_service::do_drain() {
         return bm.drain();
     });
 
-    co_await _view_builder.invoke_on_all(&db::view::view_builder::drain);
     co_await _db.invoke_on_all(&replica::database::drain);
     co_await _sys_ks.invoke_on_all(&db::system_keyspace::shutdown);
     co_await _repair.invoke_on_all(&repair_service::shutdown);
@@ -5747,7 +5739,7 @@ future<> storage_service::snitch_reconfigured() {
 
 future<raft_topology_cmd_result> storage_service::raft_topology_cmd_handler(raft::term_t term, uint64_t cmd_index, const raft_topology_cmd& cmd) {
     raft_topology_cmd_result result;
-    rtlogger.debug("topology cmd rpc {} is called", cmd.cmd);
+    rtlogger.info("topology cmd rpc {} is called index={}", cmd.cmd, cmd_index);
 
     try {
         auto& raft_server = _group0->group0_server();
@@ -5816,6 +5808,7 @@ future<raft_topology_cmd_result> storage_service::raft_topology_cmd_handler(raft
             }
             break;
             case raft_topology_cmd::command::barrier_and_drain: {
+                co_await utils::get_local_injector().inject("pause_before_barrier_and_drain", utils::wait_for_message(std::chrono::minutes(5)));
                 if (_topology_state_machine._topology.tstate == topology::transition_state::write_both_read_old) {
                     for (auto& n : _topology_state_machine._topology.transition_nodes) {
                         if (!_address_map.find(locator::host_id{n.first.uuid()})) {
@@ -6077,6 +6070,9 @@ future<raft_topology_cmd_result> storage_service::raft_topology_cmd_handler(raft
     } catch (...) {
         rtlogger.error("raft_topology_cmd {} failed with: {}", cmd.cmd, std::current_exception());
     }
+
+    rtlogger.info("topology cmd rpc {} completed with status={} index={}",
+        cmd.cmd, (result.status == raft_topology_cmd_result::command_status::success) ? "suceeded" : "failed", cmd_index);
     co_return result;
 }
 

service/storage_service.hh

@@ -48,6 +48,7 @@
 #include "timestamp.hh"
 #include "utils/user_provided_param.hh"
 #include "utils/sequenced_set.hh"
+#include "service/topology_coordinator.hh"
 
 class node_ops_cmd_request;
 class node_ops_cmd_response;
@@ -282,12 +283,12 @@ private:
     future<> snitch_reconfigured();
 
     future<mutable_token_metadata_ptr> get_mutable_token_metadata_ptr() noexcept {
-        return _shared_token_metadata.get()->clone_async().then([] (token_metadata tm) {
+        return _shared_token_metadata.get()->clone_async().then([this] (token_metadata tm) {
             // bump the token_metadata ring_version
             // to invalidate cached token/replication mappings
             // when the modified token_metadata is committed.
             tm.invalidate_cached_rings();
-            return make_ready_future<mutable_token_metadata_ptr>(make_token_metadata_ptr(std::move(tm)));
+            return _shared_token_metadata.make_token_metadata_ptr(std::move(tm));
         });
     }
 
@@ -873,6 +874,11 @@ private:
     std::optional<shared_future<>> _rebuild_result;
     std::unordered_map<raft::server_id, std::optional<shared_future<>>> _remove_result;
     tablet_op_registry _tablet_ops;
+    // This tracks active topology cmd rpc. There can be only one active
+    // cmd running and by inspecting this structure it can be checked which
+    // cmd is current executing and which nodes are still did not reply.
+    // Needed for debugging.
+    topology_coordinator_cmd_rpc_tracker _topology_cmd_rpc_tracker;
     struct {
         raft::term_t term{0};
         uint64_t last_index{0};
@@ -941,6 +947,10 @@ public:
     // Waits for topology state in which none of tablets has replaced_id as a replica.
     // Must be called on shard 0.
     future<> await_tablets_rebuilt(raft::server_id replaced_id);
+
+    topology_coordinator_cmd_rpc_tracker get_topology_cmd_status() {
+        return _topology_cmd_rpc_tracker;
+    }
 private:
     // Tracks progress of the upgrade to topology coordinator.
     future<> _upgrade_to_topology_coordinator_fiber = make_ready_future<>();

service/tablet_allocator.cc

@@ -842,7 +842,7 @@ public:
             db_clock::duration repair_time_diff;
         };
 
-        std::vector<repair_plan> plans;
+        utils::chunked_vector<repair_plan> plans;
         auto migration_tablet_ids = co_await mplan.get_migration_tablet_ids();
         for (auto&& [table, tmap_] : _tm->tablets().all_tables()) {
             auto& tmap = *tmap_;

service/topology_coordinator.cc

@@ -147,6 +147,8 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
 
     group0_voter_handler _voter_handler;
 
+    topology_coordinator_cmd_rpc_tracker& _topology_cmd_rpc_tracker;
+
     const locator::token_metadata& get_token_metadata() const noexcept {
         return *_shared_tm.get();
     }
@@ -389,6 +391,9 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
     future<> exec_direct_command_helper(raft::server_id id, uint64_t cmd_index, const raft_topology_cmd& cmd) {
         rtlogger.debug("send {} command with term {} and index {} to {}",
             cmd.cmd, _term, cmd_index, id);
+        _topology_cmd_rpc_tracker.active_dst.emplace(id);
+        auto _ = seastar::defer([this, id] { _topology_cmd_rpc_tracker.active_dst.erase(id); });
+
         auto result = _db.get_token_metadata().get_topology().is_me(to_host_id(id)) ?
                     co_await _raft_topology_cmd_handler(_term, cmd_index, cmd) :
                     co_await ser::storage_service_rpc_verbs::send_raft_topology_cmd(
@@ -403,12 +408,16 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
         auto id = node.id;
         release_node(std::move(node));
         const auto cmd_index = ++_last_cmd_index;
+        _topology_cmd_rpc_tracker.current = cmd.cmd;
+        _topology_cmd_rpc_tracker.index = cmd_index;
         co_await exec_direct_command_helper(id, cmd_index, cmd);
         co_return retake_node(co_await start_operation(), id);
     };
 
     future<> exec_global_command_helper(auto nodes, const raft_topology_cmd& cmd) {
         const auto cmd_index = ++_last_cmd_index;
+        _topology_cmd_rpc_tracker.current = cmd.cmd;
+        _topology_cmd_rpc_tracker.index = cmd_index;
         auto f = co_await coroutine::as_future(
                 seastar::parallel_for_each(std::move(nodes), [this, &cmd, cmd_index] (raft::server_id id) {
             return exec_direct_command_helper(id, cmd_index, cmd);
@@ -1510,7 +1519,13 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
                         }
                         rtlogger.info("Initiating tablet cleanup of {} on {}", gid, dst);
                         return ser::storage_service_rpc_verbs::send_tablet_cleanup(&_messaging,
-                                                                                   dst.host, _as, raft::server_id(dst.host.uuid()), gid);
+                                                                                   dst.host, _as, raft::server_id(dst.host.uuid()), gid)
+                            .then([] {
+                                return utils::get_local_injector().inject("wait_after_tablet_cleanup", [] (auto& handler) -> future<> {
+                                    rtlogger.info("Waiting after tablet cleanup");
+                                    return handler.wait_for_message(std::chrono::steady_clock::now() + std::chrono::seconds{60});
+                                });
+                            });
                     })) {
                         transition_to(locator::tablet_transition_stage::end_migration);
                     }
@@ -1730,6 +1745,11 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
     }
 
     future<> handle_tablet_resize_finalization(group0_guard g) {
+        co_await utils::get_local_injector().inject("handle_tablet_resize_finalization_wait", [] (auto& handler) -> future<> {
+            rtlogger.info("handle_tablet_resize_finalization: waiting");
+            co_await handler.wait_for_message(std::chrono::steady_clock::now() + std::chrono::seconds{60});
+        });
+
         // Executes a global barrier to guarantee that any process (e.g. repair) holding stale version
         // of token metadata will complete before we update topology.
         auto guard = co_await global_tablet_token_metadata_barrier(std::move(g));
@@ -1916,6 +1936,12 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
         }
     }
 
+    void trigger_load_stats_refresh() {
+        (void)_tablet_load_stats_refresh.trigger().handle_exception([] (auto ep) {
+            rtlogger.warn("Error during tablet load stats refresh: {}", ep);
+        });
+    }
+
     future<> cancel_all_requests(group0_guard guard, std::unordered_set<raft::server_id> dead_nodes) {
         std::vector<canonical_mutation> muts;
         std::vector<raft::server_id> reject_join;
@@ -2406,10 +2432,7 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
                     muts.emplace_back(rtbuilder.build());
                     co_await update_topology_state(take_guard(std::move(node)), std::move(muts),
                                                    "bootstrap: read fence completed");
-                    // Make sure the load balancer knows the capacity for the new node immediately.
-                    (void)_tablet_load_stats_refresh.trigger().handle_exception([] (auto ep) {
-                        rtlogger.warn("Error during tablet load stats refresh: {}", ep);
-                    });
+                    trigger_load_stats_refresh();
                     }
                     co_await _voter_handler.on_node_added(node.id, _as);
                     break;
@@ -2468,6 +2491,7 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
                     co_await db::view::view_builder::generate_mutations_on_node_left(_db, _sys_ks, node.guard.write_timestamp(), locator::host_id(replaced_node_id.uuid()), muts);
                     co_await update_topology_state(take_guard(std::move(node)), std::move(muts),
                                                   "replace: read fence completed");
+                    trigger_load_stats_refresh();
                     }
                     co_await _voter_handler.on_node_added(node.id, _as);
                     break;
@@ -2988,7 +3012,8 @@ public:
             raft_topology_cmd_handler_type raft_topology_cmd_handler,
             tablet_allocator& tablet_allocator,
             std::chrono::milliseconds ring_delay,
-            gms::feature_service& feature_service)
+            gms::feature_service& feature_service,
+            topology_coordinator_cmd_rpc_tracker& topology_cmd_rpc_tracker)
         : _sys_dist_ks(sys_dist_ks), _gossiper(gossiper), _messaging(messaging)
         , _shared_tm(shared_tm), _sys_ks(sys_ks), _db(db)
         , _group0(group0), _topo_sm(topo_sm), _as(as)
@@ -3000,6 +3025,7 @@ public:
         , _ring_delay(ring_delay)
         , _group0_holder(_group0.hold_group0_gate())
         , _voter_handler(group0, topo_sm._topology, gossiper, feature_service)
+        , _topology_cmd_rpc_tracker(topology_cmd_rpc_tracker)
         , _async_gate("topology_coordinator")
     {}
 
@@ -3614,7 +3640,8 @@ future<> run_topology_coordinator(
         tablet_allocator& tablet_allocator,
         std::chrono::milliseconds ring_delay,
         endpoint_lifecycle_notifier& lifecycle_notifier,
-        gms::feature_service& feature_service) {
+        gms::feature_service& feature_service,
+        topology_coordinator_cmd_rpc_tracker& topology_cmd_rpc_tracker) {
 
     topology_coordinator coordinator{
             sys_dist_ks, gossiper, messaging, shared_tm,
@@ -3622,7 +3649,8 @@ future<> run_topology_coordinator(
             std::move(raft_topology_cmd_handler),
             tablet_allocator,
             ring_delay,
-            feature_service};
+            feature_service,
+            topology_cmd_rpc_tracker};
 
     std::exception_ptr ex;
     lifecycle_notifier.register_subscriber(&coordinator);

service/topology_coordinator.hh

@@ -62,6 +62,12 @@ future<> wait_for_gossiper(raft::server_id id, const gms::gossiper& g, seastar::
 using raft_topology_cmd_handler_type = noncopyable_function<future<raft_topology_cmd_result>(
         raft::term_t, uint64_t, const raft_topology_cmd&)>;
 
+struct topology_coordinator_cmd_rpc_tracker {
+    raft_topology_cmd::command current;
+    uint64_t index;
+    std::set<raft::server_id> active_dst;
+};
+
 future<> run_topology_coordinator(
         seastar::sharded<db::system_distributed_keyspace>& sys_dist_ks, gms::gossiper& gossiper,
         netw::messaging_service& messaging, locator::shared_token_metadata& shared_tm,
@@ -71,6 +77,7 @@ future<> run_topology_coordinator(
         tablet_allocator& tablet_allocator,
         std::chrono::milliseconds ring_delay,
         endpoint_lifecycle_notifier& lifecycle_notifier,
-        gms::feature_service& feature_service);
+        gms::feature_service& feature_service,
+        topology_coordinator_cmd_rpc_tracker& topology_cmd_rpc_tracker);
 
 }

sstables/checksummed_data_source.cc

@@ -27,6 +27,7 @@ extern logging::logger sstlog;
 // data and no compression
 template <ChecksumUtils ChecksumType, bool check_digest>
 class checksummed_file_data_source_impl : public data_source_impl {
+    std::function<future<input_stream<char>>()> _stream_creator;
     std::optional<input_stream<char>> _input_stream;
     const checksum& _checksum;
     [[no_unique_address]] digest_members<check_digest> _digests;
@@ -38,7 +39,7 @@ class checksummed_file_data_source_impl : public data_source_impl {
     uint64_t _beg_pos;
     uint64_t _end_pos;
 public:
-    checksummed_file_data_source_impl(file f, uint64_t file_len,
+    checksummed_file_data_source_impl(stream_creator_fn stream_creator, uint64_t file_len,
                 const checksum& checksum, uint64_t pos, size_t len,
                 file_input_stream_options options,
                 std::optional<uint32_t> digest,
@@ -87,14 +88,20 @@ public:
         }
         auto start = align_down(_beg_pos, chunk_size);
         auto end = std::min(_file_len, align_up(_end_pos, chunk_size));
-        _input_stream = make_file_input_stream(std::move(f), start, end - start, std::move(options));
+        _stream_creator = [stream_creator, start, length = end - start, options] mutable {
+            return stream_creator(start, length, std::move(options));
+        };
         _underlying_pos = start;
     }
 
     virtual future<temporary_buffer<char>> get() override {
         uint64_t chunk_size = _checksum.chunk_size;
         if (_pos >= _end_pos) {
-            return make_ready_future<temporary_buffer<char>>();
+            co_return temporary_buffer<char>();
+        }
+
+        if (!_input_stream) {
+            _input_stream = co_await _stream_creator();
         }
         // Read the next chunk. We need to skip part of the first
         // chunk, but then continue to read from beginning of chunks.
@@ -103,47 +110,46 @@ public:
         if (_pos != _beg_pos && (_pos & (chunk_size - 1)) != 0) {
             throw std::runtime_error(format("Checksummed reader not aligned to chunk boundary: pos={}, chunk_size={}", _pos, chunk_size));
         }
-        return _input_stream->read_exactly(chunk_size).then([this, chunk_size](temporary_buffer<char> buf) {
-            uint32_t chunk_index = _pos >> _chunk_size_trailing_zeros;
-            if (buf.size() != chunk_size) {
-                auto actual_end = _underlying_pos + buf.size();
-                if (chunk_index + 1 < _checksum.checksums.size()) {
-                    throw malformed_sstable_exception(seastar::format("Checksummed reader hit premature end-of-file at file offset {}: expected {} chunks of size {} but data file has {}",
-                            actual_end, _checksum.checksums.size(), chunk_size, chunk_index + 1));
-                } else if (actual_end < _file_len) {
-                    // Truncation on last chunk. Update _end_pos so that future
-                    // calls to get() return immediately.
-                    _end_pos = actual_end;
-                }
-            }
-            if (chunk_index >= _checksum.checksums.size()) {
-                throw malformed_sstable_exception(seastar::format("Chunk count mismatch between CRC and Data.db: expected {} but data file has more", _checksum.checksums.size()));
-            }
-            auto expected_checksum = _checksum.checksums[chunk_index];
-            auto actual_checksum = ChecksumType::checksum(buf.get(), buf.size());
-            if (expected_checksum != actual_checksum) {
-                _error_handler(seastar::format(
-                        "Checksummed chunk of size {} at file offset {} failed checksum: expected={}, actual={}",
-                        buf.size(), _underlying_pos, expected_checksum, actual_checksum));
+        auto buf = co_await _input_stream->read_exactly(chunk_size);
+        uint32_t chunk_index = _pos >> _chunk_size_trailing_zeros;
+        if (buf.size() != chunk_size) {
+            auto actual_end = _underlying_pos + buf.size();
+            if (chunk_index + 1 < _checksum.checksums.size()) {
+                throw malformed_sstable_exception(seastar::format("Checksummed reader hit premature end-of-file at file offset {}: expected {} chunks of size {} but data file has {}",
+                        actual_end, _checksum.checksums.size(), chunk_size, chunk_index + 1));
+            } else if (actual_end < _file_len) {
+                // Truncation on last chunk. Update _end_pos so that future
+                // calls to get() return immediately.
+                _end_pos = actual_end;
             }
+        }
+        if (chunk_index >= _checksum.checksums.size()) {
+            throw malformed_sstable_exception(seastar::format("Chunk count mismatch between CRC and Data.db: expected {} but data file has more", _checksum.checksums.size()));
+        }
+        auto expected_checksum = _checksum.checksums[chunk_index];
+        auto actual_checksum = ChecksumType::checksum(buf.get(), buf.size());
+        if (expected_checksum != actual_checksum) {
+            _error_handler(seastar::format(
+                    "Checksummed chunk of size {} at file offset {} failed checksum: expected={}, actual={}",
+                    buf.size(), _underlying_pos, expected_checksum, actual_checksum));
+        }
 
-            if constexpr (check_digest) {
-                if (_digests.can_calculate_digest) {
-                    _digests.actual_digest = checksum_combine_or_feed<ChecksumType>(_digests.actual_digest, actual_checksum, buf.begin(), buf.size());
-                }
+        if constexpr (check_digest) {
+            if (_digests.can_calculate_digest) {
+                _digests.actual_digest = checksum_combine_or_feed<ChecksumType>(_digests.actual_digest, actual_checksum, buf.begin(), buf.size());
             }
+        }
 
-            buf.trim_front(_pos & (chunk_size - 1));
-            _pos += buf.size();
-            _underlying_pos += chunk_size;
+        buf.trim_front(_pos & (chunk_size - 1));
+        _pos += buf.size();
+        _underlying_pos += chunk_size;
 
-            if constexpr (check_digest) {
-                if (_digests.can_calculate_digest && _pos == _file_len && _digests.expected_digest != _digests.actual_digest) {
-                    _error_handler(seastar::format("Digest mismatch: expected={}, actual={}", _digests.expected_digest, _digests.actual_digest));
-                }
+        if constexpr (check_digest) {
+            if (_digests.can_calculate_digest && _pos == _file_len && _digests.expected_digest != _digests.actual_digest) {
+                _error_handler(seastar::format("Digest mismatch: expected={}, actual={}", _digests.expected_digest, _digests.actual_digest));
             }
-            return buf;
-        });
+        }
+        co_return buf;
     }
 
     virtual future<> close() override {
@@ -166,59 +172,61 @@ public:
         }
         _pos += n;
         if (_pos == _end_pos) {
-            return make_ready_future<temporary_buffer<char>>();
+            co_return temporary_buffer<char>();
         }
         auto underlying_n = align_down(_pos, chunk_size) - _underlying_pos;
         _beg_pos = _pos;
         _underlying_pos += underlying_n;
-        return _input_stream->skip(underlying_n).then([] {
-            return make_ready_future<temporary_buffer<char>>();
-        });
+        if (!_input_stream) {
+            _input_stream = co_await _stream_creator();
+        }
+        co_await _input_stream->skip(underlying_n);
+        co_return temporary_buffer<char>();
     }
 };
 
 template <ChecksumUtils ChecksumType, bool check_digest>
 class checksummed_file_data_source : public data_source {
 public:
-    checksummed_file_data_source(file f, uint64_t file_len, const checksum& checksum,
+    checksummed_file_data_source(stream_creator_fn stream_creator, uint64_t file_len, const checksum& checksum,
             uint64_t offset, size_t len, file_input_stream_options options,
             std::optional<uint32_t> digest, integrity_error_handler error_handler)
         : data_source(std::make_unique<checksummed_file_data_source_impl<ChecksumType, check_digest>>(
-                std::move(f), file_len, checksum, offset, len, std::move(options), digest,
+                std::move(stream_creator), file_len, checksum, offset, len, std::move(options), digest,
                 error_handler))
     {}
 };
 
 template <ChecksumUtils ChecksumType>
 inline input_stream<char> make_checksummed_file_input_stream(
-        file f, uint64_t file_len, const checksum& checksum, uint64_t offset,
+        stream_creator_fn stream_creator, uint64_t file_len, const checksum& checksum, uint64_t offset,
         size_t len, file_input_stream_options options, std::optional<uint32_t> digest,
         integrity_error_handler error_handler)
 {
     if (digest) {
         return input_stream<char>(checksummed_file_data_source<ChecksumType, true>(
-            std::move(f), file_len, checksum, offset, len, std::move(options), digest,
+            std::move(stream_creator), file_len, checksum, offset, len, std::move(options), digest,
             error_handler));
     }
     return input_stream<char>(checksummed_file_data_source<ChecksumType, false>(
-        std::move(f), file_len, checksum, offset, len, std::move(options), digest, error_handler));
+        std::move(stream_creator), file_len, checksum, offset, len, std::move(options), digest, error_handler));
 }
 
 input_stream<char> make_checksummed_file_k_l_format_input_stream(
-        file f, uint64_t file_len, const checksum& checksum, uint64_t offset,
+        stream_creator_fn stream_creator, uint64_t file_len, const checksum& checksum, uint64_t offset,
         size_t len, file_input_stream_options options, std::optional<uint32_t> digest,
         integrity_error_handler error_handler)
 {
-    return make_checksummed_file_input_stream<adler32_utils>(std::move(f), file_len,
+    return make_checksummed_file_input_stream<adler32_utils>(std::move(stream_creator), file_len,
             checksum, offset, len, std::move(options), digest, error_handler);
 }
 
 input_stream<char> make_checksummed_file_m_format_input_stream(
-        file f, uint64_t file_len, const checksum& checksum, uint64_t offset,
+        stream_creator_fn stream_creator, uint64_t file_len, const checksum& checksum, uint64_t offset,
         size_t len, file_input_stream_options options, std::optional<uint32_t> digest,
         integrity_error_handler error_handler)
 {
-    return make_checksummed_file_input_stream<crc32_utils>(std::move(f), file_len,
+    return make_checksummed_file_input_stream<crc32_utils>(std::move(stream_creator), file_len,
             checksum, offset, len, std::move(options), digest, error_handler);
 }
 

sstables/checksummed_data_source.hh

@@ -9,24 +9,26 @@
 #pragma once
 
 #include <seastar/core/seastar.hh>
+#include <seastar/core/fstream.hh>
 #include <seastar/core/iostream.hh>
 
 #include "sstables/types.hh"
 
 namespace sstables {
 
+using stream_creator_fn = std::function<future<input_stream<char>>(uint64_t, uint64_t, file_input_stream_options)>;
 using integrity_error_handler = std::function<void(sstring)>;
 
 void throwing_integrity_error_handler(sstring msg);
 
-input_stream<char> make_checksummed_file_k_l_format_input_stream(file f,
+input_stream<char> make_checksummed_file_k_l_format_input_stream(stream_creator_fn stream_creator,
                 uint64_t file_len, const sstables::checksum& checksum,
                 uint64_t offset, size_t len,
                 class file_input_stream_options options,
                 std::optional<uint32_t> digest,
                 integrity_error_handler error_handler = throwing_integrity_error_handler);
 
-input_stream<char> make_checksummed_file_m_format_input_stream(file f,
+input_stream<char> make_checksummed_file_m_format_input_stream(stream_creator_fn stream_creator,
                 uint64_t file_len, const sstables::checksum& checksum,
                 uint64_t offset, size_t len,
                 class file_input_stream_options options,

sstables/compress.cc

@@ -304,6 +304,7 @@ enum class compressed_checksum_mode {
 
 template <ChecksumUtils ChecksumType, bool check_digest, compressed_checksum_mode mode>
 class compressed_file_data_source_impl : public data_source_impl {
+    std::function<future<input_stream<char>>()> _stream_creator;
     std::optional<input_stream<char>> _input_stream;
     sstables::compression* _compression_metadata;
     sstables::compression::segmented_offsets::accessor _offsets;
@@ -314,7 +315,7 @@ class compressed_file_data_source_impl : public data_source_impl {
     uint64_t _beg_pos;
     uint64_t _end_pos;
 public:
-    compressed_file_data_source_impl(file f, sstables::compression* cm,
+    compressed_file_data_source_impl(sstables::stream_creator_fn stream_creator, sstables::compression* cm,
                 uint64_t pos, size_t len, file_input_stream_options options,
                 reader_permit permit, std::optional<uint32_t> digest)
             : _compression_metadata(cm)
@@ -352,15 +353,18 @@ public:
         // and open a file_input_stream to read that range.
         auto start = _compression_metadata->locate(_beg_pos, _offsets);
         auto end = _compression_metadata->locate(_end_pos - 1, _offsets);
-        _input_stream = make_file_input_stream(std::move(f),
-                start.chunk_start,
-                end.chunk_start + end.chunk_len - start.chunk_start,
-                std::move(options));
+        _stream_creator = [stream_creator{std::move(stream_creator)}, start = start.chunk_start, length = end.chunk_start + end.chunk_len - start.chunk_start, options] mutable {
+            return stream_creator(start, length, std::move(options));
+        };
         _underlying_pos = start.chunk_start;
     }
     virtual future<temporary_buffer<char>> get() override {
         if (_pos >= _end_pos) {
-            return make_ready_future<temporary_buffer<char>>();
+            co_return temporary_buffer<char>();
+        }
+
+        if (!_input_stream) {
+            _input_stream = co_await _stream_creator();
         }
         auto addr = _compression_metadata->locate(_pos, _offsets);
         // Uncompress the next chunk. We need to skip part of the first
@@ -371,58 +375,55 @@ public:
         if (!addr.chunk_len) {
             throw sstables::malformed_sstable_exception(format("compressed chunk_len must be greater than zero, chunk_start={}", addr.chunk_start));
         }
-        return _input_stream->read_exactly(addr.chunk_len).then([this, addr](temporary_buffer<char> buf) {
-            if (buf.size() != addr.chunk_len) {
-                throw sstables::malformed_sstable_exception(format("compressed reader hit premature end-of-file at file offset {}, expected chunk_len={}, actual={}", _underlying_pos, addr.chunk_len, buf.size()));
+        auto buf = co_await _input_stream->read_exactly(addr.chunk_len);
+        if (buf.size() != addr.chunk_len) {
+            throw sstables::malformed_sstable_exception(format("compressed reader hit premature end-of-file at file offset {}, expected chunk_len={}, actual={}", _underlying_pos, addr.chunk_len, buf.size()));
+        }
+        auto res_units = co_await _permit.request_memory(_compression_metadata->uncompressed_chunk_length());
+        // The last 4 bytes of the chunk are the adler32/crc32 checksum
+        // of the rest of the (compressed) chunk.
+        auto compressed_len = addr.chunk_len - 4;
+        // FIXME: Do not always calculate checksum - Cassandra has a
+        // probability (defaulting to 1.0, but still...)
+        auto expected_checksum = read_be<uint32_t>(buf.get() + compressed_len);
+        auto actual_checksum = ChecksumType::checksum(buf.get(), compressed_len);
+        if (expected_checksum != actual_checksum) {
+            throw sstables::malformed_sstable_exception(format("compressed chunk of size {} at file offset {} failed checksum, expected={}, actual={}", addr.chunk_len, _underlying_pos, expected_checksum, actual_checksum));
+        }
+
+        if constexpr (check_digest) {
+            if (_digests.can_calculate_digest) {
+                _digests.actual_digest = checksum_combine_or_feed<ChecksumType>(_digests.actual_digest, actual_checksum, buf.get(), compressed_len);
+                if constexpr (mode == compressed_checksum_mode::checksum_all) {
+                    uint32_t be_actual_checksum = cpu_to_be(actual_checksum);
+                    _digests.actual_digest = ChecksumType::checksum(_digests.actual_digest,
+                            reinterpret_cast<const char*>(&be_actual_checksum), sizeof(be_actual_checksum));
+                }
             }
-            return _permit.request_memory(_compression_metadata->uncompressed_chunk_length()).then(
-                    [this, addr, buf = std::move(buf)] (reader_permit::resource_units res_units) mutable {
-                // The last 4 bytes of the chunk are the adler32/crc32 checksum
-                // of the rest of the (compressed) chunk.
-                auto compressed_len = addr.chunk_len - 4;
-                // FIXME: Do not always calculate checksum - Cassandra has a
-                // probability (defaulting to 1.0, but still...)
-                auto expected_checksum = read_be<uint32_t>(buf.get() + compressed_len);
-                auto actual_checksum = ChecksumType::checksum(buf.get(), compressed_len);
-                if (expected_checksum != actual_checksum) {
-                    throw sstables::malformed_sstable_exception(format("compressed chunk of size {} at file offset {} failed checksum, expected={}, actual={}", addr.chunk_len, _underlying_pos, expected_checksum, actual_checksum));
-                }
+        }
 
-                if constexpr (check_digest) {
-                    if (_digests.can_calculate_digest) {
-                        _digests.actual_digest = checksum_combine_or_feed<ChecksumType>(_digests.actual_digest, actual_checksum, buf.get(), compressed_len);
-                        if constexpr (mode == compressed_checksum_mode::checksum_all) {
-                            uint32_t be_actual_checksum = cpu_to_be(actual_checksum);
-                            _digests.actual_digest = ChecksumType::checksum(_digests.actual_digest,
-                                    reinterpret_cast<const char*>(&be_actual_checksum), sizeof(be_actual_checksum));
-                        }
-                    }
-                }
+        // We know that the uncompressed data will take exactly
+        // chunk_length bytes (or less, if reading the last chunk).
+        temporary_buffer<char> out(
+                _compression_metadata->uncompressed_chunk_length());
+        // The compressed data is the whole chunk, minus the last 4
+        // bytes (which contain the checksum verified above).
 
-                // We know that the uncompressed data will take exactly
-                // chunk_length bytes (or less, if reading the last chunk).
-                temporary_buffer<char> out(
-                        _compression_metadata->uncompressed_chunk_length());
-                // The compressed data is the whole chunk, minus the last 4
-                // bytes (which contain the checksum verified above).
+        auto len = _compression_metadata->get_compressor().uncompress(buf.get(), compressed_len, out.get_write(), out.size());
 
-                auto len = _compression_metadata->get_compressor().uncompress(buf.get(), compressed_len, out.get_write(), out.size());
+        out.trim(len);
+        out.trim_front(addr.offset);
+        _pos += out.size();
+        _underlying_pos += addr.chunk_len;
 
-                out.trim(len);
-                out.trim_front(addr.offset);
-                _pos += out.size();
-                _underlying_pos += addr.chunk_len;
-
-                if constexpr (check_digest) {
-                    if (_digests.can_calculate_digest
-                            && _pos == _compression_metadata->uncompressed_file_length()
-                            && _digests.expected_digest != _digests.actual_digest) {
-                        throw sstables::malformed_sstable_exception(seastar::format("Digest mismatch: expected={}, actual={}", _digests.expected_digest, _digests.actual_digest));
-                    }
-                }
-                return make_tracked_temporary_buffer(std::move(out), std::move(res_units));
-            });
-        });
+        if constexpr (check_digest) {
+            if (_digests.can_calculate_digest
+                    && _pos == _compression_metadata->uncompressed_file_length()
+                    && _digests.expected_digest != _digests.actual_digest) {
+                throw sstables::malformed_sstable_exception(seastar::format("Digest mismatch: expected={}, actual={}", _digests.expected_digest, _digests.actual_digest));
+            }
+        }
+        co_return make_tracked_temporary_buffer(std::move(out), std::move(res_units));
     }
 
     virtual future<> close() override {
@@ -444,41 +445,42 @@ public:
         }
         _pos += n;
         if (_pos == _end_pos) {
-            return make_ready_future<temporary_buffer<char>>();
+            co_return temporary_buffer<char>();
         }
         auto addr = _compression_metadata->locate(_pos, _offsets);
         auto underlying_n = addr.chunk_start - _underlying_pos;
         _underlying_pos = addr.chunk_start;
         _beg_pos = _pos;
-        return _input_stream->skip(underlying_n).then([] {
-            return make_ready_future<temporary_buffer<char>>();
-        });
+        if (!_input_stream) {
+            _input_stream = co_await _stream_creator();
+        }
+        co_await _input_stream->skip(underlying_n);
+        co_return temporary_buffer<char>();
     }
 };
 
 template <ChecksumUtils ChecksumType, bool check_digest, compressed_checksum_mode mode>
 class compressed_file_data_source : public data_source {
 public:
-    compressed_file_data_source(file f, sstables::compression* cm,
+    compressed_file_data_source(sstables::stream_creator_fn stream_creator, sstables::compression* cm,
             uint64_t offset, size_t len, file_input_stream_options options, reader_permit permit,
             std::optional<uint32_t> digest)
         : data_source(std::make_unique<compressed_file_data_source_impl<ChecksumType, check_digest, mode>>(
-                std::move(f), cm, offset, len, std::move(options), std::move(permit), digest))
+                std::move(stream_creator), cm, offset, len, std::move(options), std::move(permit), digest))
         {}
 };
 
 template <ChecksumUtils ChecksumType, compressed_checksum_mode mode>
-inline input_stream<char> make_compressed_file_input_stream(
-        file f, sstables::compression *cm, uint64_t offset, size_t len,
+inline input_stream<char> make_compressed_file_input_stream(sstables::stream_creator_fn stream_creator, sstables::compression *cm, uint64_t offset, size_t len,
         file_input_stream_options options, reader_permit permit,
         std::optional<uint32_t> digest)
 {
     if (digest) [[unlikely]] {
         return input_stream<char>(compressed_file_data_source<ChecksumType, true, mode>(
-                std::move(f), cm, offset, len, std::move(options), std::move(permit), digest));
+                std::move(stream_creator), cm, offset, len, std::move(options), std::move(permit), digest));
     }
     return input_stream<char>(compressed_file_data_source<ChecksumType, false, mode>(
-            std::move(f), cm, offset, len, std::move(options), std::move(permit), digest));
+            std::move(stream_creator), cm, offset, len, std::move(options), std::move(permit), digest));
 }
 
 // compressed_file_data_sink_impl works as a filter for a file output stream,
@@ -577,21 +579,21 @@ inline output_stream<char> make_compressed_file_output_stream(output_stream<char
     return output_stream<char>(compressed_file_data_sink<ChecksumType, mode>(std::move(out), cm));
 }
 
-input_stream<char> sstables::make_compressed_file_k_l_format_input_stream(file f,
+input_stream<char> sstables::make_compressed_file_k_l_format_input_stream(stream_creator_fn stream_creator,
         sstables::compression* cm, uint64_t offset, size_t len,
         class file_input_stream_options options, reader_permit permit,
         std::optional<uint32_t> digest)
 {
     return make_compressed_file_input_stream<adler32_utils, compressed_checksum_mode::checksum_chunks_only>(
-            std::move(f), cm, offset, len, std::move(options), std::move(permit), digest);
+            std::move(stream_creator), cm, offset, len, std::move(options), std::move(permit), digest);
 }
 
-input_stream<char> sstables::make_compressed_file_m_format_input_stream(file f,
+input_stream<char> sstables::make_compressed_file_m_format_input_stream(stream_creator_fn stream_creator,
         sstables::compression *cm, uint64_t offset, size_t len,
         class file_input_stream_options options, reader_permit permit,
         std::optional<uint32_t> digest) {
     return make_compressed_file_input_stream<crc32_utils, compressed_checksum_mode::checksum_all>(
-            std::move(f), cm, offset, len, std::move(options), std::move(permit), digest);
+            std::move(stream_creator), cm, offset, len, std::move(options), std::move(permit), digest);
 }
 
 output_stream<char> sstables::make_compressed_file_m_format_output_stream(output_stream<char> out,

sstables/compress.hh

@@ -361,17 +361,19 @@ public:
     friend class sstable;
 };
 
+using stream_creator_fn = std::function<future<input_stream<char>>(uint64_t, uint64_t, file_input_stream_options)>;
+
 // Note: compression_metadata is passed by reference; The caller is
 // responsible for keeping the compression_metadata alive as long as there
 // are open streams on it. This should happen naturally on a higher level -
 // as long as we have *sstables* work in progress, we need to keep the whole
 // sstable alive, and the compression metadata is only a part of it.
-input_stream<char> make_compressed_file_k_l_format_input_stream(file f,
+input_stream<char> make_compressed_file_k_l_format_input_stream(stream_creator_fn stream_creator,
                 sstables::compression* cm, uint64_t offset, size_t len,
                 class file_input_stream_options options, reader_permit permit,
                 std::optional<uint32_t> digest);
 
-input_stream<char> make_compressed_file_m_format_input_stream(file f,
+input_stream<char> make_compressed_file_m_format_input_stream(stream_creator_fn stream_creator,
                 sstables::compression* cm, uint64_t offset, size_t len,
                 class file_input_stream_options options, reader_permit permit,
                 std::optional<uint32_t> digest);

sstables/index_reader.hh

@@ -454,12 +454,12 @@ class index_reader {
     bool _single_page_read;
     abort_source _abort;
 
-    std::unique_ptr<index_consume_entry_context<index_consumer>> make_context(uint64_t begin, uint64_t end, index_consumer& consumer) {
+    future<std::unique_ptr<index_consume_entry_context<index_consumer>>> make_context(uint64_t begin, uint64_t end, index_consumer& consumer) {
         auto index_file = make_tracked_index_file(*_sstable, _permit, _trace_state, _use_caching);
-        auto input = make_file_input_stream(index_file, begin, (_single_page_read ? end : _sstable->index_size()) - begin,
-                        get_file_input_stream_options());
+        auto input = input_stream<char>(co_await _sstable->get_storage().make_data_or_index_source(
+            *_sstable, component_type::Index, index_file, begin, (_single_page_read ? end : _sstable->index_size()) - begin, get_file_input_stream_options()));
         auto trust_pi = trust_promoted_index(_sstable->has_correct_promoted_index_entries());
-        return std::make_unique<index_consume_entry_context<index_consumer>>(*_sstable, _permit, consumer, trust_pi, std::move(input),
+        co_return std::make_unique<index_consume_entry_context<index_consumer>>(*_sstable, _permit, consumer, trust_pi, std::move(input),
                             begin, end - begin, _sstable->get_column_translation(), _abort, _trace_state);
     }
 
@@ -467,12 +467,12 @@ class index_reader {
         assert(!bound.context || !_single_page_read);
         if (!bound.context) {
             bound.consumer = std::make_unique<index_consumer>(_region, _sstable->get_schema());
-            bound.context = make_context(begin, end, *bound.consumer);
+            bound.context = co_await make_context(begin, end, *bound.consumer);
             bound.consumer->prepare(quantity);
-            return make_ready_future<>();
+            co_return;
         }
         bound.consumer->prepare(quantity);
-        return bound.context->fast_forward_to(begin, end);
+        co_return co_await bound.context->fast_forward_to(begin, end);
     }
 
 private:

sstables/kl/reader.cc

@@ -1343,12 +1343,12 @@ private:
 
         if (_single_partition_read) {
             _read_enabled = (begin != *end);
-            _context = data_consume_single_partition<DataConsumeRowsContext>(*_schema, _sst, _consumer, { begin, *end }, integrity_check::no);
+            _context = co_await data_consume_single_partition<DataConsumeRowsContext>(*_schema, _sst, _consumer, { begin, *end }, integrity_check::no);
         } else {
             sstable::disk_read_range drr{begin, *end};
             auto last_end = _fwd_mr ? _sst->data_size() : drr.end;
             _read_enabled = bool(drr);
-            _context = data_consume_rows<DataConsumeRowsContext>(*_schema, _sst, _consumer, std::move(drr), last_end, integrity_check::no);
+            _context = co_await data_consume_rows<DataConsumeRowsContext>(*_schema, _sst, _consumer, std::move(drr), last_end, integrity_check::no);
         }
 
         _monitor.on_read_started(_context->reader_position());
@@ -1545,6 +1545,7 @@ class sstable_full_scan_reader : public mp_row_consumer_reader_k_l {
     Consumer _consumer;
     std::unique_ptr<DataConsumeRowsContext> _context;
     read_monitor& _monitor;
+    integrity_check _integrity_check;
 public:
     sstable_full_scan_reader(shared_sstable sst, schema_ptr schema,
              reader_permit permit,
@@ -1553,9 +1554,8 @@ public:
              integrity_check integrity)
         : mp_row_consumer_reader_k_l(std::move(schema), permit, std::move(sst))
         , _consumer(this, _schema, std::move(permit), _schema->full_slice(), std::move(trace_state), streamed_mutation::forwarding::no, _sst)
-        , _context(data_consume_rows<DataConsumeRowsContext>(*_schema, _sst, _consumer, integrity))
-        , _monitor(mon) {
-        _monitor.on_read_started(_context->reader_position());
+        , _monitor(mon)
+        , _integrity_check(integrity) {
     }
 public:
     void on_out_of_clustering_range() override {
@@ -1571,14 +1571,18 @@ public:
         on_internal_error(sstlog, "sstable_full_scan_reader: doesn't support next_partition()");
     }
     virtual future<> fill_buffer() override {
+        if (!_context) {
+            _context = co_await data_consume_rows<DataConsumeRowsContext>(*_schema, _sst, _consumer, _integrity_check);
+            _monitor.on_read_started(_context->reader_position());
+        }
         if (_end_of_stream) {
-            return make_ready_future<>();
+            co_return;
         }
         if (_context->eof()) {
             _end_of_stream = true;
-            return make_ready_future<>();
+            co_return;
         }
-        return _context->consume_input();
+        co_return co_await _context->consume_input();
     }
     virtual future<> close() noexcept override {
         if (!_context) {

sstables/mx/partition_reversing_data_source.cc

@@ -402,7 +402,7 @@ class partition_reversing_data_source_impl final : public data_source_impl {
         FINISHED
     } _state = state::RANGE_END;
 private:
-    input_stream<char> data_stream(size_t start, size_t end) {
+    future<input_stream<char>> data_stream(size_t start, size_t end) {
         return _sst->data_stream(start, end - start, _permit, _trace_state, {});
     }
     future<temporary_buffer<char>> data_read(uint64_t start, uint64_t end) {
@@ -474,7 +474,7 @@ public:
 
     virtual future<temporary_buffer<char>> get() override {
         if (!_partition_header_context) {
-            _partition_header_context.emplace(data_stream(_partition_start, _partition_end), _partition_start, _partition_end - _partition_start, _permit);
+            _partition_header_context.emplace(co_await data_stream(_partition_start, _partition_end), _partition_start, _partition_end - _partition_start, _permit);
             co_await _partition_header_context->consume_input();
             _clustering_range_start = _partition_header_context->header_end_pos();
             co_return co_await data_read(_partition_start, _clustering_range_start);
@@ -507,7 +507,7 @@ public:
                 }
                 look_in_last_block = true;
             } else {
-                co_await emplace_row_skipping_context(data_stream(_row_start, _row_end), _row_start, _row_end);
+                co_await emplace_row_skipping_context(co_await data_stream(_row_start, _row_end), _row_start, _row_end);
                 co_await _row_skipping_context->consume_input();
                 if (_row_skipping_context->end_of_partition()) {
                     look_in_last_block = true;
@@ -526,7 +526,7 @@ public:
                     _row_start = _clustering_range_start;
                 }
                 uint64_t last_row_start = _row_start;
-                co_await emplace_row_skipping_context(data_stream(_row_start, _partition_end), _row_start, _partition_end);
+                co_await emplace_row_skipping_context(co_await data_stream(_row_start, _partition_end), _row_start, _partition_end);
                 co_await _row_skipping_context->consume_input();
                 while (!_row_skipping_context->end_of_partition()) {
                     last_row_start = _row_start;

sstables/mx/reader.cc

@@ -1568,13 +1568,13 @@ private:
                 _context = std::move(reversed_context.the_context);
                 _reversed_read_sstable_position = &reversed_context.current_position_in_sstable;
             } else {
-                _context = data_consume_single_partition<DataConsumeRowsContext>(*_schema, _sst, _consumer, { begin, *end }, _integrity);
+                _context = co_await data_consume_single_partition<DataConsumeRowsContext>(*_schema, _sst, _consumer, { begin, *end }, _integrity);
             }
         } else {
             sstable::disk_read_range drr{begin, *end};
             auto last_end = _fwd_mr ? _sst->data_size() : drr.end;
             _read_enabled = bool(drr);
-            _context = data_consume_rows<DataConsumeRowsContext>(*_schema, _sst, _consumer, std::move(drr), last_end, _integrity);
+            _context = co_await data_consume_rows<DataConsumeRowsContext>(*_schema, _sst, _consumer, std::move(drr), last_end, _integrity);
         }
 
         _monitor.on_read_started(_context->reader_position());
@@ -1813,7 +1813,7 @@ private:
             _checksum = co_await _sst->read_checksum();
             co_await _sst->read_digest();
         }
-        _context = data_consume_rows<DataConsumeRowsContext>(*_schema, _sst, _consumer, _integrity);
+        _context = co_await data_consume_rows<DataConsumeRowsContext>(*_schema, _sst, _consumer, _integrity);
         _monitor.on_read_started(_context->reader_position());
     }
 public:
@@ -2105,7 +2105,7 @@ future<uint64_t> validate(
         sstables::read_monitor& monitor) {
     auto schema = sstable->get_schema();
     validating_consumer consumer(schema, permit, sstable, std::move(error_handler));
-    auto context = data_consume_rows<data_consume_rows_context_m<validating_consumer>>(*schema, sstable, consumer, integrity_check::yes);
+    auto context = co_await data_consume_rows<data_consume_rows_context_m<validating_consumer>>(*schema, sstable, consumer, integrity_check::yes);
 
     std::optional<sstables::index_reader> idx_reader;
     idx_reader.emplace(sstable, permit, tracing::trace_state_ptr{}, sstables::use_caching::no, false);

sstables/sstable_mutation_reader.hh

@@ -109,16 +109,16 @@ position_in_partition_view get_slice_lower_bound(const schema& s, const query::p
 // The amount of this excessive read is controlled by read ahead
 // heuristics which learn from the usefulness of previous read aheads.
 template <typename DataConsumeRowsContext>
-inline std::unique_ptr<DataConsumeRowsContext> data_consume_rows(const schema& s, shared_sstable sst, typename DataConsumeRowsContext::consumer& consumer,
+inline future<std::unique_ptr<DataConsumeRowsContext>> data_consume_rows(const schema& s, shared_sstable sst, typename DataConsumeRowsContext::consumer& consumer,
         sstable::disk_read_range toread, uint64_t last_end, integrity_check integrity) {
     // Although we were only asked to read until toread.end, we'll not limit
     // the underlying file input stream to this end, but rather to last_end.
     // This potentially enables read-ahead beyond end, until last_end, which
     // can be beneficial if the user wants to fast_forward_to() on the
     // returned context, and may make small skips.
-    auto input = sst->data_stream(toread.start, last_end - toread.start,
+    auto input = co_await sst->data_stream(toread.start, last_end - toread.start,
             consumer.permit(), consumer.trace_state(), sst->_partition_range_history, sstable::raw_stream::no, integrity);
-    return std::make_unique<DataConsumeRowsContext>(s, std::move(sst), consumer, std::move(input), toread.start, toread.end - toread.start);
+    co_return std::make_unique<DataConsumeRowsContext>(s, std::move(sst), consumer, std::move(input), toread.start, toread.end - toread.start);
 }
 
 template <typename DataConsumeRowsContext>
@@ -149,16 +149,16 @@ inline reversed_context<DataConsumeRowsContext> data_consume_reversed_partition(
 }
 
 template <typename DataConsumeRowsContext>
-inline std::unique_ptr<DataConsumeRowsContext> data_consume_single_partition(const schema& s, shared_sstable sst, typename DataConsumeRowsContext::consumer& consumer,
+inline future<std::unique_ptr<DataConsumeRowsContext>> data_consume_single_partition(const schema& s, shared_sstable sst, typename DataConsumeRowsContext::consumer& consumer,
         sstable::disk_read_range toread, integrity_check integrity) {
-    auto input = sst->data_stream(toread.start, toread.end - toread.start,
+    auto input = co_await sst->data_stream(toread.start, toread.end - toread.start,
             consumer.permit(), consumer.trace_state(), sst->_single_partition_history, sstable::raw_stream::no, integrity);
-    return std::make_unique<DataConsumeRowsContext>(s, std::move(sst), consumer, std::move(input), toread.start, toread.end - toread.start);
+    co_return std::make_unique<DataConsumeRowsContext>(s, std::move(sst), consumer, std::move(input), toread.start, toread.end - toread.start);
 }
 
 // Like data_consume_rows() with bounds, but iterates over whole range
 template <typename DataConsumeRowsContext>
-inline std::unique_ptr<DataConsumeRowsContext> data_consume_rows(const schema& s, shared_sstable sst, typename DataConsumeRowsContext::consumer& consumer,
+inline future<std::unique_ptr<DataConsumeRowsContext>> data_consume_rows(const schema& s, shared_sstable sst, typename DataConsumeRowsContext::consumer& consumer,
         integrity_check integrity) {
     auto data_size = sst->data_size();
     return data_consume_rows<DataConsumeRowsContext>(s, std::move(sst), consumer, {0, data_size}, data_size, integrity);

sstables/sstables.cc

@@ -2458,7 +2458,7 @@ component_type sstable::component_from_sstring(version_types v, const sstring &s
     }
 }
 
-input_stream<char> sstable::data_stream(uint64_t pos, size_t len,
+future<input_stream<char>> sstable::data_stream(uint64_t pos, size_t len,
         reader_permit permit, tracing::trace_state_ptr trace_state, lw_shared_ptr<file_input_stream_history> history, raw_stream raw,
         integrity_check integrity, integrity_error_handler error_handler) {
     file_input_stream_options options;
@@ -2475,13 +2475,15 @@ input_stream<char> sstable::data_stream(uint64_t pos, size_t len,
     if (integrity == integrity_check::yes) {
         digest = get_digest();
     }
-
+    auto stream_creator = [this, f](uint64_t pos, uint64_t len, file_input_stream_options options) mutable -> future<input_stream<char>> {
+        co_return input_stream<char>(co_await _storage->make_data_or_index_source(*this, component_type::Data, std::move(f), pos, len, std::move(options)));
+    };
     if (_components->compression && raw == raw_stream::no) {
         if (_version >= sstable_version_types::mc) {
-            return make_compressed_file_m_format_input_stream(f, &_components->compression,
+            co_return make_compressed_file_m_format_input_stream(stream_creator, &_components->compression,
                pos, len, std::move(options), permit, digest);
         } else {
-            return make_compressed_file_k_l_format_input_stream(f, &_components->compression,
+            co_return make_compressed_file_k_l_format_input_stream(stream_creator, &_components->compression,
                 pos, len, std::move(options), permit, digest);
         }
     }
@@ -2489,22 +2491,21 @@ input_stream<char> sstable::data_stream(uint64_t pos, size_t len,
         auto checksum = get_checksum();
         auto file_len = data_size();
         if (_version >= sstable_version_types::mc) {
-             return make_checksummed_file_m_format_input_stream(f, file_len,
+             co_return make_checksummed_file_m_format_input_stream(stream_creator, file_len,
                 *checksum, pos, len, std::move(options), digest, error_handler);
         } else {
-            return make_checksummed_file_k_l_format_input_stream(f, file_len,
+            co_return make_checksummed_file_k_l_format_input_stream(stream_creator, file_len,
                 *checksum, pos, len, std::move(options), digest, error_handler);
         }
     }
-    return make_file_input_stream(f, pos, len, std::move(options));
+    co_return co_await stream_creator(pos, len, std::move(options));
 }
 
 future<temporary_buffer<char>> sstable::data_read(uint64_t pos, size_t len, reader_permit permit) {
-    return do_with(data_stream(pos, len, std::move(permit), tracing::trace_state_ptr(), {}), [len] (auto& stream) {
-        return stream.read_exactly(len).finally([&stream] {
-            return stream.close();
-        });
-    });
+    auto stream = co_await data_stream(pos, len, std::move(permit), tracing::trace_state_ptr(), {});
+    auto buff = co_await stream.read_exactly(len);
+    co_await stream.close();
+    co_return buff;
 }
 
 template <typename ChecksumType>
@@ -2670,10 +2671,10 @@ future<validate_checksums_result> validate_checksums(shared_sstable sst, reader_
 
     input_stream<char> data_stream;
     if (sst->get_compression()) {
-        data_stream = sst->data_stream(0, sst->ondisk_data_size(), permit,
+        data_stream = co_await sst->data_stream(0, sst->ondisk_data_size(), permit,
                 nullptr, nullptr, sstable::raw_stream::yes);
     } else {
-        data_stream = sst->data_stream(0, sst->data_size(), permit,
+        data_stream = co_await sst->data_stream(0, sst->data_size(), permit,
                 nullptr, nullptr, sstable::raw_stream::no,
                 integrity_check::yes, [&ret](sstring msg) {
                     sstlog.error("{}", msg);
@@ -3654,6 +3655,9 @@ future<data_sink> file_io_extension::wrap_sink(const sstable& sst, component_typ
     co_return co_await make_file_data_sink(std::move(f), file_output_stream_options{});
 }
 
+future<data_source> file_io_extension::wrap_source(const sstable& sst, component_type c, sstables::data_source_creator_fn, uint64_t, uint64_t) {
+    SCYLLA_ASSERT(0 && "You are not supposed to get here, file_io_extension::wrap_source() is not implemented");
+}
 } // namespace sstables
 
 namespace seastar {

sstables/sstables.hh

@@ -752,7 +752,7 @@ public:
     // integrity-checked stream with no compression. The parameter is ignored
     // if integrity checking is disabled or the SSTable is compressed.
     using raw_stream = bool_class<class raw_stream_tag>;
-    input_stream<char> data_stream(uint64_t pos, size_t len,
+    future<input_stream<char>> data_stream(uint64_t pos, size_t len,
             reader_permit permit, tracing::trace_state_ptr trace_state, lw_shared_ptr<file_input_stream_history> history,
             raw_stream raw = raw_stream::no, integrity_check integrity = integrity_check::no,
             integrity_error_handler error_handler = throwing_integrity_error_handler);
@@ -1051,13 +1051,13 @@ public:
     friend class promoted_index;
     friend class sstables_manager;
     template <typename DataConsumeRowsContext>
-    friend std::unique_ptr<DataConsumeRowsContext>
+    friend future<std::unique_ptr<DataConsumeRowsContext>>
     data_consume_rows(const schema&, shared_sstable, typename DataConsumeRowsContext::consumer&, disk_read_range, uint64_t, integrity_check);
     template <typename DataConsumeRowsContext>
-    friend std::unique_ptr<DataConsumeRowsContext>
+    friend future<std::unique_ptr<DataConsumeRowsContext>>
     data_consume_single_partition(const schema&, shared_sstable, typename DataConsumeRowsContext::consumer&, disk_read_range, integrity_check);
     template <typename DataConsumeRowsContext>
-    friend std::unique_ptr<DataConsumeRowsContext>
+    friend future<std::unique_ptr<DataConsumeRowsContext>>
     data_consume_rows(const schema&, shared_sstable, typename DataConsumeRowsContext::consumer&, integrity_check);
     friend void lw_shared_ptr_deleter<sstables::sstable>::dispose(sstable* s);
     gc_clock::time_point get_gc_before_for_drop_estimation(const gc_clock::time_point& compaction_time, const tombstone_gc_state& gc_state, const schema_ptr& s) const;
@@ -1126,6 +1126,8 @@ public:
     // output device. Default impl will call wrap_file and generate a wrapper object.
     virtual future<data_sink> wrap_sink(const sstable&, component_type, data_sink);
 
+    virtual future<data_source>
+    wrap_source(const sstable&, component_type, sstables::data_source_creator_fn, uint64_t offset, uint64_t len);
     // optionally return a map of attributes for a given sstable,
     // suitable for "describe".
     // This would preferably be interesting info on what/why the extension did

sstables/storage.cc

@@ -89,6 +89,7 @@ public:
     virtual future<> wipe(const sstable& sst, sync_dir) noexcept override;
     virtual future<file> open_component(const sstable& sst, component_type type, open_flags flags, file_open_options options, bool check_integrity) override;
     virtual future<data_sink> make_data_or_index_sink(sstable& sst, component_type type) override;
+    future<data_source> make_data_or_index_source(sstable& sst, component_type type, file f, uint64_t offset, uint64_t len, file_input_stream_options opt) const override;
     virtual future<data_sink> make_component_sink(sstable& sst, component_type type, open_flags oflags, file_output_stream_options options) override;
     virtual future<> destroy(const sstable& sst) override { return make_ready_future<>(); }
     virtual future<atomic_delete_context> atomic_delete_prepare(const std::vector<shared_sstable>&) const override;
@@ -110,6 +111,11 @@ future<data_sink> filesystem_storage::make_data_or_index_sink(sstable& sst, comp
     return make_file_data_sink(type == component_type::Data ? std::move(sst._data_file) : std::move(sst._index_file), options);
 }
 
+future<data_source> filesystem_storage::make_data_or_index_source(sstable&, component_type type, file f, uint64_t offset, uint64_t len, file_input_stream_options opt) const {
+    SCYLLA_ASSERT(type == component_type::Data || type == component_type::Index);
+    co_return make_file_data_source(std::move(f), offset, len, std::move(opt));
+}
+
 future<data_sink> filesystem_storage::make_component_sink(sstable& sst, component_type type, open_flags oflags, file_output_stream_options options) {
     return sst.new_sstable_component_file(sst._write_error_handler, type, oflags).then([options = std::move(options)] (file f) mutable {
         return make_file_data_sink(std::move(f), std::move(options));
@@ -570,6 +576,7 @@ public:
     virtual future<> wipe(const sstable& sst, sync_dir) noexcept override;
     virtual future<file> open_component(const sstable& sst, component_type type, open_flags flags, file_open_options options, bool check_integrity) override;
     virtual future<data_sink> make_data_or_index_sink(sstable& sst, component_type type) override;
+    future<data_source> make_data_or_index_source(sstable& sst, component_type type, file f, uint64_t offset, uint64_t len, file_input_stream_options opt) const override;
     virtual future<data_sink> make_component_sink(sstable& sst, component_type type, open_flags oflags, file_output_stream_options options) override;
     virtual future<> destroy(const sstable& sst) override {
         return make_ready_future<>();
@@ -639,12 +646,45 @@ static future<data_sink> maybe_wrap_sink(const sstable& sst, component_type type
     co_return sink;
 }
 
+static future<data_source> maybe_wrap_source(const sstable& sst, component_type type, data_source_creator_fn source_creator, uint64_t offset, uint64_t len) {
+    if (type != component_type::TOC && type != component_type::TemporaryTOC) {
+        for (auto* ext : sst.manager().config().extensions().sstable_file_io_extensions()) {
+            std::exception_ptr p;
+            try {
+                co_return co_await ext->wrap_source(sst, type, std::move(source_creator), offset, len);
+            } catch (...) {
+                p = std::current_exception();
+            }
+            if (p) {
+                std::rethrow_exception(std::move(p));
+            }
+        }
+    }
+    co_return source_creator(offset, len);
+}
+
 future<data_sink> s3_storage::make_data_or_index_sink(sstable& sst, component_type type) {
     SCYLLA_ASSERT(type == component_type::Data || type == component_type::Index);
     // FIXME: if we have file size upper bound upfront, it's better to use make_upload_sink() instead
     return maybe_wrap_sink(sst, type, _client->make_upload_jumbo_sink(make_s3_object_name(sst, type), std::nullopt, _as));
 }
 
+future<data_source>
+s3_storage::make_data_or_index_source(sstable& sst, component_type type, file f, uint64_t offset, uint64_t len, file_input_stream_options options) const {
+    if (offset == 0) {
+        co_return co_await maybe_wrap_source(
+            sst,
+            type,
+            [this, object_name = make_s3_object_name(sst, type)](uint64_t, uint64_t) {
+                return _client->make_chunked_download_source(object_name, s3::full_range, _as);
+            },
+            offset,
+            len);
+    }
+    co_return make_file_data_source(
+        co_await maybe_wrap_file(sst, type, open_flags::ro, _client->make_readable_file(make_s3_object_name(sst, type), _as)), offset, len, std::move(options));
+}
+
 future<data_sink> s3_storage::make_component_sink(sstable& sst, component_type type, open_flags oflags, file_output_stream_options options) {
     return maybe_wrap_sink(sst, type, _client->make_upload_sink(make_s3_object_name(sst, type), _as));
 }

sstables/storage.hh

@@ -106,6 +106,7 @@ public:
     virtual future<> wipe(const sstable& sst, sync_dir) noexcept = 0;
     virtual future<file> open_component(const sstable& sst, component_type type, open_flags flags, file_open_options options, bool check_integrity) = 0;
     virtual future<data_sink> make_data_or_index_sink(sstable& sst, component_type type) = 0;
+    virtual future<data_source> make_data_or_index_source(sstable& sst, component_type type, file f, uint64_t offset, uint64_t len, file_input_stream_options opt) const = 0;
     virtual future<data_sink> make_component_sink(sstable& sst, component_type type, open_flags oflags, file_output_stream_options options) = 0;
     virtual future<> destroy(const sstable& sst) = 0;
     virtual future<atomic_delete_context> atomic_delete_prepare(const std::vector<shared_sstable>&) const = 0;
@@ -124,4 +125,5 @@ future<> init_keyspace_storage(const sstables_manager&, const data_dictionary::s
 
 std::vector<std::filesystem::path> get_local_directories(const db::config& db, const data_dictionary::storage_options::local& so);
 
+using data_source_creator_fn = std::function<data_source(uint64_t, uint64_t)>;
 } // namespace sstables

streaming/consumer.cc

@@ -7,6 +7,7 @@
  */
 
 #include <seastar/core/coroutine.hh>
+#include <seastar/core/with_scheduling_group.hh>
 
 #include "consumer.hh"
 #include "replica/database.hh"
@@ -35,7 +36,9 @@ mutation_reader_consumer make_streaming_consumer(sstring origin,
 
             auto cf = db.local().find_column_family(reader.schema()).shared_from_this();
             auto guard = service::topology_guard(frozen_guard);
-            auto use_view_update_path = co_await db::view::check_needs_view_update_path(vb.local(), db.local().get_token_metadata_ptr(), *cf, reason);
+            bool use_view_update_path = co_await with_scheduling_group(db.local().get_gossip_scheduling_group(), [&] {
+                return db::view::check_needs_view_update_path(vb.local(), db.local().get_token_metadata_ptr(), *cf, reason);
+            });
             //FIXME: for better estimations this should be transmitted from remote
             auto metadata = mutation_source_metadata{};
             auto& cs = cf->get_compaction_strategy();

streaming/stream_session.cc

@@ -264,13 +264,14 @@ void stream_manager::init_messaging_service_handler(abort_source& as) {
                     make_generating_reader_v1(s, permit, std::move(get_next_mutation_fragment)),
                     make_streaming_consumer(estimated_partitions, reason, topo_guard),
                     std::move(op)
-                ).then_wrapped(std::ref(result_handling)).handle_exception([s, plan_id, from, sink] (std::exception_ptr ep) {
+                ).then_wrapped(std::ref(result_handling)).handle_exception([s, plan_id, from, sink] (std::exception_ptr ep) mutable -> future<> {
                     auto level = seastar::log_level::error;
                     if (try_catch<seastar::rpc::closed_error>(ep)) {
                         level = seastar::log_level::debug;
                     }
                     sslog.log(level, "[Stream #{}] Failed to handle STREAM_MUTATION_FRAGMENTS (respond phase) for ks={}, cf={}, peer={}: {}",
                             plan_id, s->ks_name(), s->cf_name(), from, ep);
+                    co_await sink.close();
                 });
             });
           } catch (...) {

tasks/task_manager.cc

@@ -121,10 +121,6 @@ future<std::optional<double>> task_manager::task::impl::expected_total_workload(
     return make_ready_future<std::optional<double>>(std::nullopt);
 }
 
-std::optional<double> task_manager::task::impl::expected_children_number() const {
-    return std::nullopt;
-}
-
 task_manager::task::progress task_manager::task::impl::get_binary_progress() const {
     return tasks::task_manager::task::progress{
         .completed = is_complete(),
@@ -133,20 +129,10 @@ task_manager::task::progress task_manager::task::impl::get_binary_progress() con
 }
 
 future<task_manager::task::progress> task_manager::task::impl::get_progress() const {
-    auto children_num = _children.size();
-    if (children_num == 0) {
-        co_return get_binary_progress();
+    std::optional<double> expected_workload = co_await expected_total_workload();
+    if (!expected_workload && _children.size() == 0) {
+        co_return task_manager::task::progress{};
     }
-
-    std::optional<double> expected_workload = std::nullopt;
-    auto expected_children_num = expected_children_number();
-    // When get_progress is called, the task can have some of its children unregistered yet.
-    // Then if total workload is not known, progress obtained from children may be deceiving.
-    // In such a situation it's safer to return binary progress value.
-    if (expected_children_num.value_or(0) != children_num && !(expected_workload = co_await expected_total_workload())) {
-        co_return get_binary_progress();
-    }
-
     auto progress = co_await _children.get_progress(_status.progress_units);
     progress.total = expected_workload.value_or(progress.total);
     co_return progress;