replica: Remove unnecessary noexcept

Can potentially lead to unnecessary abort.

compaction_groups() and for_each_compaction_group() can throw.

Co-authored-by: bhalevy <20910904+bhalevy@users.noreply.github.com>

.github/scripts/auto-backport.py

@@ -62,7 +62,7 @@ def create_pull_request(repo, new_branch_name, base_branch_name, pr, backport_pr
         if is_draft:
             labels_to_add.append("conflicts")
             pr_comment = f"@{pr.user.login} - This PR was marked as draft because it has conflicts\n"
-            pr_comment += "Please resolve them and remove the 'conflicts' label. The PR will be made ready for review automatically."
+            pr_comment += "Please resolve them and mark this PR as ready for review"
             backport_pr.create_issue_comment(pr_comment)
         
         # Apply all labels at once if we have any

auth/service.cc

@@ -884,6 +884,15 @@ future<> migrate_to_auth_v2(db::system_keyspace& sys_ks, ::service::raft_group0_
             ::service::client_state cs(::service::client_state::internal_tag{}, tc);
             ::service::query_state qs(cs, empty_service_permit());
 
+            auto rows = co_await qp.execute_internal(
+                    seastar::format("SELECT * FROM {}.{}", meta::legacy::AUTH_KS, cf_name),
+                    db::consistency_level::ALL,
+                    qs,
+                    {},
+                    cql3::query_processor::cache_internal::no);
+            if (rows->empty()) {
+                continue;
+            }
             std::vector<sstring> col_names;
             for (const auto& col : schema->all_columns()) {
                 col_names.push_back(col.name_as_cql_string());
@@ -892,14 +901,7 @@ future<> migrate_to_auth_v2(db::system_keyspace& sys_ks, ::service::raft_group0_
             for (size_t i = 1; i < col_names.size(); ++i) {
                 val_binders_str += ", ?";
             }
-
-            co_await qp.query_internal(
-                    seastar::format("SELECT * FROM {}.{}", meta::legacy::AUTH_KS, cf_name),
-                    db::consistency_level::ALL,
-                    qs,
-                    {},
-                    1000,
-                    [&](const cql3::untyped_result_set_row& row) -> future<stop_iteration> {
+            for (const auto& row : *rows) {
                 std::vector<data_value_or_unset> values;
                 for (const auto& col : schema->all_columns()) {
                     if (row.has(col.name_as_text())) {
@@ -923,8 +925,7 @@ future<> migrate_to_auth_v2(db::system_keyspace& sys_ks, ::service::raft_group0_
                             format("expecting single insert mutation, got {}", muts.size()));
                 }
                 co_yield std::move(muts[0]);
-                co_return stop_iteration::no;
-            });
+            }
         }
         co_yield co_await sys_ks.make_auth_version_mutation(ts,
                 db::system_keyspace::auth_version_t::v2);
@@ -933,7 +934,7 @@ future<> migrate_to_auth_v2(db::system_keyspace& sys_ks, ::service::raft_group0_
             start_operation_func,
             std::move(gen),
             as,
-            get_raft_timeout());
+            std::nullopt);
 }
 
 }

cql3/query_processor.cc

@@ -886,10 +886,9 @@ future<> query_processor::for_each_cql_result(
 }
 
 future<::shared_ptr<untyped_result_set>>
-query_processor::execute_paged_internal(internal_query_state& state, service::query_state* query_state) {
+query_processor::execute_paged_internal(internal_query_state& state) {
     state.p->statement->validate(*this, service::client_state::for_internal_calls());
-    auto default_qs = query_state ? std::nullopt : std::make_optional(query_state_for_internal_call());
-    auto& qs = query_state ? *query_state : *default_qs;
+    auto qs = query_state_for_internal_call();
     ::shared_ptr<cql_transport::messages::result_message> msg =
       co_await state.p->statement->execute(*this, qs, *state.opts, std::nullopt);
 
@@ -926,20 +925,6 @@ query_processor::execute_paged_internal(internal_query_state& state, service::qu
     co_return ::make_shared<untyped_result_set>(msg);
 }
 
-future<> query_processor::for_each_cql_result(
-        cql3::internal_query_state& state,
-        service::query_state& query_state,
-        noncopyable_function<future<stop_iteration>(const cql3::untyped_result_set::row&)> f) {
-    do {
-        auto msg = co_await execute_paged_internal(state, &query_state);
-        for (auto& row : *msg) {
-            if ((co_await f(row)) == stop_iteration::yes) {
-                co_return;
-            }
-        }
-    } while (has_more_results(state));
-}
-
 future<::shared_ptr<untyped_result_set>>
 query_processor::execute_internal(
         const sstring& query_string,
@@ -1217,17 +1202,6 @@ future<> query_processor::query_internal(
     co_return co_await for_each_cql_result(query_state, std::move(f));
 }
 
-future<> query_processor::query_internal(
-        const sstring& query_string,
-        db::consistency_level cl,
-        service::query_state& query_state,
-        const data_value_list& values,
-        int32_t page_size,
-        noncopyable_function<future<stop_iteration>(const cql3::untyped_result_set_row&)> f) {
-    auto paged_state = create_paged_state(query_string, cl, values, page_size);
-    co_return co_await for_each_cql_result(paged_state, query_state, std::move(f));
-}
-
 future<> query_processor::query_internal(
         const sstring& query_string,
         noncopyable_function<future<stop_iteration>(const cql3::untyped_result_set_row&)> f) {

cql3/query_processor.hh

@@ -332,29 +332,6 @@ public:
             int32_t page_size,
             noncopyable_function<future<stop_iteration>(const cql3::untyped_result_set_row&)> f);
 
-    /*!
-     * \brief iterate over all cql results using paging with a custom query_state (for timeout control)
-     *
-     * You can use placeholders in the query, the statement will only be prepared once.
-     *
-     * query_string - the cql string, can contain placeholders
-     * cl - consistency level of the query
-     * query_state - query state with custom timeout configuration
-     * values - values to be substituted for the placeholders in the query
-     * page_size - maximum page size
-     * f - a function to be run on each row of the query result,
-     *     if the function returns stop_iteration::yes the iteration will stop
-     *
-     * \note This function is optimized for convenience, not performance.
-     */
-    future<> query_internal(
-            const sstring& query_string,
-            db::consistency_level cl,
-            service::query_state& query_state,
-            const data_value_list& values,
-            int32_t page_size,
-            noncopyable_function<future<stop_iteration>(const cql3::untyped_result_set_row&)> f);
-
     /*
      * \brief iterate over all cql results using paging
      * An overload of query_internal without query parameters
@@ -524,14 +501,11 @@ private:
             int32_t page_size);
 
     /*!
-     * \brief run a query using paging with an optional custom query_state (for timeout control)
-     *
-     * state - internal query state containing prepared statement and options
-     * query_state - optional query state with custom timeout configuration (defaults to internal query state)
+     * \brief run a query using paging
      *
      * \note Optimized for convenience, not performance.
      */
-    future<::shared_ptr<untyped_result_set>> execute_paged_internal(internal_query_state& state, service::query_state* query_state = nullptr);
+    future<::shared_ptr<untyped_result_set>> execute_paged_internal(internal_query_state& state);
 
     /*!
      * \brief iterate over all results using paging, accept a function that returns a future
@@ -542,21 +516,6 @@ private:
             cql3::internal_query_state& state,
             noncopyable_function<future<stop_iteration>(const cql3::untyped_result_set_row&)> f);
 
-    /*!
-     * \brief iterate over all results using paging with a custom query_state (for timeout control)
-     *
-     * state - internal query state containing prepared statement and options
-     * query_state - query state with custom timeout configuration
-     * f - a function to be run on each row of the query result,
-     *     if the function returns stop_iteration::yes the iteration will stop
-     *
-     * \note Optimized for convenience, not performance.
-     */
-    future<> for_each_cql_result(
-            cql3::internal_query_state& state,
-            service::query_state& query_state,
-            noncopyable_function<future<stop_iteration>(const cql3::untyped_result_set_row&)> f);
-
     /*!
      * \brief check, based on the state if there are additional results
      * Users of the paging, should not use the internal_query_state directly

replica/compaction_group.hh

@@ -297,17 +297,17 @@ public:
 
     const dht::token_range& token_range() const noexcept;
 
-    size_t memtable_count() const noexcept;
+    size_t memtable_count() const;
 
     const compaction_group_ptr& main_compaction_group() const noexcept;
     const std::vector<compaction_group_ptr>& split_ready_compaction_groups() const;
     compaction_group_ptr& select_compaction_group(locator::tablet_range_side) noexcept;
 
-    uint64_t live_disk_space_used() const noexcept;
+    uint64_t live_disk_space_used() const;
 
-    void for_each_compaction_group(std::function<void(const compaction_group_ptr&)> action) const noexcept;
-    utils::small_vector<compaction_group_ptr, 3> compaction_groups() noexcept;
-    utils::small_vector<const_compaction_group_ptr, 3> compaction_groups() const noexcept;
+    void for_each_compaction_group(std::function<void(const compaction_group_ptr&)> action) const;
+    utils::small_vector<compaction_group_ptr, 3> compaction_groups();
+    utils::small_vector<const_compaction_group_ptr, 3> compaction_groups() const;
 
     utils::small_vector<compaction_group_ptr, 3> split_unready_groups() const;
     bool split_unready_groups_are_empty() const;
@@ -430,7 +430,7 @@ public:
     virtual storage_group& storage_group_for_token(dht::token) const = 0;
     virtual utils::chunked_vector<storage_group_ptr> storage_groups_for_token_range(dht::token_range tr) const = 0;
 
-    virtual locator::combined_load_stats table_load_stats(std::function<bool(const locator::tablet_map&, locator::global_tablet_id)> tablet_filter) const noexcept = 0;
+    virtual locator::combined_load_stats table_load_stats(std::function<bool(const locator::tablet_map&, locator::global_tablet_id)> tablet_filter) const = 0;
     virtual bool all_storage_groups_split() = 0;
     virtual future<> split_all_storage_groups(tasks::task_info tablet_split_task_info) = 0;
     virtual future<> maybe_split_compaction_group_of(size_t idx) = 0;

replica/database.hh

@@ -1133,7 +1133,7 @@ public:
 
     // The tablet filter is used to not double account migrating tablets, so it's important that
     // only one of pending or leaving replica is accounted based on current migration stage.
-    locator::combined_load_stats table_load_stats(std::function<bool(const locator::tablet_map&, locator::global_tablet_id)> tablet_filter) const noexcept;
+    locator::combined_load_stats table_load_stats(std::function<bool(const locator::tablet_map&, locator::global_tablet_id)> tablet_filter) const;
 
     const db::view::stats& get_view_stats() const {
         return _view_stats;

replica/table.cc

@@ -708,7 +708,7 @@ public:
         return *_single_sg;
     }
 
-    locator::combined_load_stats table_load_stats(std::function<bool(const locator::tablet_map&, locator::global_tablet_id)>) const noexcept override {
+    locator::combined_load_stats table_load_stats(std::function<bool(const locator::tablet_map&, locator::global_tablet_id)>) const override {
         return locator::combined_load_stats{
             .table_ls = locator::table_load_stats{
                             .size_in_bytes = _single_sg->live_disk_space_used(),
@@ -874,7 +874,7 @@ public:
         return storage_group_for_id(storage_group_of(token).first);
     }
 
-    locator::combined_load_stats table_load_stats(std::function<bool(const locator::tablet_map&, locator::global_tablet_id)> tablet_filter) const noexcept override;
+    locator::combined_load_stats table_load_stats(std::function<bool(const locator::tablet_map&, locator::global_tablet_id)> tablet_filter) const override;
     bool all_storage_groups_split() override;
     future<> split_all_storage_groups(tasks::task_info tablet_split_task_info) override;
     future<> maybe_split_compaction_group_of(size_t idx) override;
@@ -922,7 +922,7 @@ compaction_group_ptr& storage_group::select_compaction_group(locator::tablet_ran
     return _main_cg;
 }
 
-void storage_group::for_each_compaction_group(std::function<void(const compaction_group_ptr&)> action) const noexcept {
+void storage_group::for_each_compaction_group(std::function<void(const compaction_group_ptr&)> action) const {
     action(_main_cg);
     for (auto& cg : _merging_groups) {
         action(cg);
@@ -932,7 +932,7 @@ void storage_group::for_each_compaction_group(std::function<void(const compactio
     }
 }
 
-utils::small_vector<compaction_group_ptr, 3> storage_group::compaction_groups() noexcept {
+utils::small_vector<compaction_group_ptr, 3> storage_group::compaction_groups() {
     utils::small_vector<compaction_group_ptr, 3> cgs;
     for_each_compaction_group([&cgs] (const compaction_group_ptr& cg) {
         cgs.push_back(cg);
@@ -940,7 +940,7 @@ utils::small_vector<compaction_group_ptr, 3> storage_group::compaction_groups()
     return cgs;
 }
 
-utils::small_vector<const_compaction_group_ptr, 3> storage_group::compaction_groups() const noexcept {
+utils::small_vector<const_compaction_group_ptr, 3> storage_group::compaction_groups() const {
     utils::small_vector<const_compaction_group_ptr, 3> cgs;
     for_each_compaction_group([&cgs] (const compaction_group_ptr& cg) {
         cgs.push_back(cg);
@@ -1890,7 +1890,7 @@ sstables::file_size_stats compaction_group::live_disk_space_used_full_stats() co
     return _main_sstables->get_file_size_stats() + _maintenance_sstables->get_file_size_stats();
 }
 
-uint64_t storage_group::live_disk_space_used() const noexcept {
+uint64_t storage_group::live_disk_space_used() const {
     auto cgs = const_cast<storage_group&>(*this).compaction_groups();
     return std::ranges::fold_left(cgs | std::views::transform(std::mem_fn(&compaction_group::live_disk_space_used)), uint64_t(0), std::plus{});
 }
@@ -2813,7 +2813,7 @@ void table::on_flush_timer() {
     });
 }
 
-locator::combined_load_stats tablet_storage_group_manager::table_load_stats(std::function<bool(const locator::tablet_map&, locator::global_tablet_id)> tablet_filter) const noexcept {
+locator::combined_load_stats tablet_storage_group_manager::table_load_stats(std::function<bool(const locator::tablet_map&, locator::global_tablet_id)> tablet_filter) const {
     locator::table_load_stats table_stats;
     table_stats.split_ready_seq_number = _split_ready_seq_number;
 
@@ -2836,7 +2836,7 @@ locator::combined_load_stats tablet_storage_group_manager::table_load_stats(std:
     };
 }
 
-locator::combined_load_stats table::table_load_stats(std::function<bool(const locator::tablet_map&, locator::global_tablet_id)> tablet_filter) const noexcept {
+locator::combined_load_stats table::table_load_stats(std::function<bool(const locator::tablet_map&, locator::global_tablet_id)> tablet_filter) const {
     return _sg_manager->table_load_stats(std::move(tablet_filter));
 }
 
@@ -3456,7 +3456,7 @@ size_t compaction_group::memtable_count() const noexcept {
     return _memtables->size();
 }
 
-size_t storage_group::memtable_count() const noexcept {
+size_t storage_group::memtable_count() const {
     return std::ranges::fold_left(compaction_groups() | std::views::transform(std::mem_fn(&compaction_group::memtable_count)), size_t(0), std::plus{});
 }
 

test/cqlpy/nodetool.py

@@ -67,11 +67,11 @@ nodetool_cmd.conf = False
 
 # Run the external "nodetool" executable (can be overridden by the NODETOOL
 # environment variable). Only call this if the REST API doesn't work.
-def run_nodetool(cql, *args, **subprocess_kwargs):
+def run_nodetool(cql, *args):
     # TODO: We may need to change this function or its callers to add proper
     # support for testing on multi-node clusters.
     host = cql.cluster.contact_points[0]
-    return subprocess.run([nodetool_cmd(), '-h', host, *args], **subprocess_kwargs)
+    subprocess.run([nodetool_cmd(), '-h', host, *args])
 
 def flush(cql, table):
     ks, cf = table.split('.')
@@ -159,28 +159,6 @@ def disablebinary(cql):
     else:
         run_nodetool(cql, "disablebinary")
 
-def getlogginglevel(cql, logger):
-    if has_rest_api(cql):
-        resp = requests.get(f'{rest_api_url(cql)}/system/logger/{logger}')
-        if resp.ok:
-            return resp.text.strip()
-        raise RuntimeError(f"failed to fetch logging level for {logger}: {resp.status_code} {resp.text}")
-
-    result = run_nodetool(
-        cql,
-        "getlogginglevels",
-        capture_output=True,
-        text=True,
-        check=True,
-    )
-    for line in result.stdout.splitlines():
-        stripped = line.strip()
-        parts = stripped.split()
-        if len(parts) >= 2 and parts[0] == logger:
-            return parts[-1]
-
-    raise RuntimeError(f"logger {logger} not found in getlogginglevels output")
-
 def setlogginglevel(cql, logger, level):
     if has_rest_api(cql):
         requests.post(f'{rest_api_url(cql)}/system/logger/{logger}', params={'level': level})

test/cqlpy/test_protocol_exceptions.py

@@ -10,7 +10,6 @@ import re
 import requests
 import socket
 import struct
-from test.cqlpy import nodetool
 from test.cqlpy.util import cql_session
 
 def get_protocol_error_metrics(host) -> int:
@@ -59,50 +58,11 @@ def try_connect(host, port, creds, protocol_version):
     with cql_with_protocol(host, port, creds, protocol_version) as session:
         return 1 if session else 0
 
-@pytest.fixture
-def debug_exceptions_logging(request, cql):
-    def _read_level() -> str | None:
-        try:
-            level = nodetool.getlogginglevel(cql, "exception")
-            if level:
-                level = level.strip().strip('"').lower()
-            return level
-        except Exception as exc:
-            print(f"Failed to read exception logger level: {exc}")
-            return None
-
-    def _set_and_verify(level: str) -> bool:
-        try:
-            nodetool.setlogginglevel(cql, "exception", level)
-        except Exception as exc:
-            print(f"Failed to set exception logger level to '{level}': {exc}")
-            return False
-
-        observed = _read_level()
-        if observed == level:
-            return True
-
-        print(f"Exception logger level observed as '{observed}' while expecting '{level}'")
-        return False
-
-    def _restore_logging():
-        if not enabled and previous_level is None:
-            return
-
-        target_level = previous_level or "info"
-        _set_and_verify(target_level)
-
-    previous_level = _read_level()
-    enabled = _set_and_verify("debug")
-
-    yield
-    _restore_logging()
-
 # If there is a protocol version mismatch, the server should
 # raise a protocol error, which is counted in the metrics.
-def test_protocol_version_mismatch(scylla_only, debug_exceptions_logging, request, host):
-    run_count = 200
-    cpp_exception_threshold = 20
+def test_protocol_version_mismatch(scylla_only, request, host):
+    run_count = 100
+    cpp_exception_threshold = 10
 
     cpp_exception_metrics_before = get_cpp_exceptions_metrics(host)
     protocol_exception_metrics_before = get_protocol_error_metrics(host)
@@ -284,8 +244,8 @@ def _protocol_error_impl(
         s.close()
 
 def _test_impl(host, flag):
-    run_count = 200
-    cpp_exception_threshold = 20
+    run_count = 100
+    cpp_exception_threshold = 10
 
     cpp_exception_metrics_before = get_cpp_exceptions_metrics(host)
     protocol_exception_metrics_before = get_protocol_error_metrics(host)
@@ -307,47 +267,47 @@ def no_ssl(request):
     yield
 
 # Malformed BATCH with an invalid kind triggers a protocol error.
-def test_invalid_kind_in_batch_message(scylla_only, no_ssl, debug_exceptions_logging, host):
+def test_invalid_kind_in_batch_message(scylla_only, no_ssl, host):
     _test_impl(host, "trigger_bad_batch")
 
 # Send OPTIONS during AUTHENTICATE to trigger auth-state error.
-def test_unexpected_message_during_auth(scylla_only, no_ssl, debug_exceptions_logging, host):
+def test_unexpected_message_during_auth(scylla_only, no_ssl, host):
     _test_impl(host, "trigger_unexpected_auth")
 
 # STARTUP with an invalid/missing string-map entry should produce a protocol error.
-def test_process_startup_invalid_string_map(scylla_only, no_ssl, debug_exceptions_logging, host):
+def test_process_startup_invalid_string_map(scylla_only, no_ssl, host):
     _test_impl(host, "trigger_process_startup_invalid_string_map")
 
 # STARTUP with unknown COMPRESSION option should produce a protocol error.
-def test_unknown_compression_algorithm(scylla_only, no_ssl, debug_exceptions_logging, host):
+def test_unknown_compression_algorithm(scylla_only, no_ssl, host):
     _test_impl(host, "trigger_unknown_compression")
 
 # QUERY long-string truncation: declared length > provided bytes triggers protocol error.
-def test_process_query_internal_malformed_query(scylla_only, no_ssl, debug_exceptions_logging, host):
+def test_process_query_internal_malformed_query(scylla_only, no_ssl, host):
     _test_impl(host, "trigger_process_query_internal_malformed_query")
 
 # QUERY options malformed: PAGE_SIZE flag set but page_size truncated triggers protocol error.
-def test_process_query_internal_fail_read_options(scylla_only, no_ssl, debug_exceptions_logging, host):
+def test_process_query_internal_fail_read_options(scylla_only, no_ssl, host):
     _test_impl(host, "trigger_process_query_internal_fail_read_options")
 
 # PREPARE long-string truncation: declared length > provided bytes triggers protocol error.
-def test_process_prepare_malformed_query(scylla_only, no_ssl, debug_exceptions_logging, host):
+def test_process_prepare_malformed_query(scylla_only, no_ssl, host):
     _test_impl(host, "trigger_process_prepare_malformed_query")
 
 # EXECUTE cache-key malformed: short-bytes length > provided bytes triggers protocol error.
-def test_process_execute_internal_malformed_cache_key(scylla_only, no_ssl, debug_exceptions_logging, host):
+def test_process_execute_internal_malformed_cache_key(scylla_only, no_ssl, host):
     _test_impl(host, "trigger_process_execute_internal_malformed_cache_key")
 
 # REGISTER malformed string list: declared string length > provided bytes triggers protocol error.
-def test_process_register_malformed_string_list(scylla_only, no_ssl, debug_exceptions_logging, host):
+def test_process_register_malformed_string_list(scylla_only, no_ssl, host):
     _test_impl(host, "trigger_process_register_malformed_string_list")
 
 # Test if the protocol exceptions do not decrease after running the test happy path.
 # This is to ensure that the protocol exceptions are not cleared or reset
 # during the test execution.
-def test_no_protocol_exceptions(scylla_only, no_ssl, debug_exceptions_logging, host):
-    run_count = 200
-    cpp_exception_threshold = 20
+def test_no_protocol_exceptions(scylla_only, no_ssl, host):
+    run_count = 100
+    cpp_exception_threshold = 10
 
     cpp_exception_metrics_before = get_cpp_exceptions_metrics(host)
     protocol_exception_metrics_before = get_protocol_error_metrics(host)