Merge "No infinite time-outs for internal distributed queries" from J…

…esse

"
This series replaces infinite time-outs in internal distributed
(non-local) CQL queries with finite ones.

The implementation of tracing, which also performs internal queries,
already has finite time-outs, so it is unchanged.

Fixes scylladb#3603.
"

* 'jhk/finite_time_outs/v2' of https://github.com/hakuch/scylla:
  Use finite time-outs for internal auth. queries
  Use finite query time-outs for `system_distributed`

auth/common.cc

@@ -28,6 +28,7 @@
 #include "database.hh"
 #include "schema_builder.hh"
 #include "service/migration_manager.hh"
+#include "timeout_config.hh"
 
 namespace auth {
 
@@ -94,4 +95,10 @@ future<> wait_for_schema_agreement(::service::migration_manager& mm, const datab
     });
 }
 
+const timeout_config& internal_distributed_timeout_config() noexcept {
+    static const auto t = 5s;
+    static const timeout_config tc{t, t, t, t, t, t, t};
+    return tc;
+}
+
 }

auth/common.hh

@@ -38,6 +38,7 @@
 using namespace std::chrono_literals;
 
 class database;
+class timeout_config;
 
 namespace service {
 class migration_manager;
@@ -82,4 +83,9 @@ future<> create_metadata_table_if_missing(
 
 future<> wait_for_schema_agreement(::service::migration_manager&, const database&);
 
+///
+/// Time-outs for internal, non-local CQL queries.
+///
+const timeout_config& internal_distributed_timeout_config() noexcept;
+
 }

auth/default_authorizer.cc

@@ -228,7 +228,7 @@ default_authorizer::modify(
         return _qp.process(
                 query,
                 db::consistency_level::ONE,
-                infinite_timeout_config,
+                internal_distributed_timeout_config(),
                 {permissions::to_strings(set), sstring(role_name), resource.name()}).discard_result();
     });
 }
@@ -254,7 +254,7 @@ future<std::vector<permission_details>> default_authorizer::list_all() const {
     return _qp.process(
             query,
             db::consistency_level::ONE,
-            infinite_timeout_config,
+            internal_distributed_timeout_config(),
             {},
             true).then([](::shared_ptr<cql3::untyped_result_set> results) {
         std::vector<permission_details> all_details;
@@ -282,7 +282,7 @@ future<> default_authorizer::revoke_all(stdx::string_view role_name) const {
     return _qp.process(
             query,
             db::consistency_level::ONE,
-            infinite_timeout_config,
+            internal_distributed_timeout_config(),
             {sstring(role_name)}).discard_result().handle_exception([role_name](auto ep) {
         try {
             std::rethrow_exception(ep);

auth/password_authenticator.cc

@@ -183,15 +183,15 @@ future<> password_authenticator::migrate_legacy_metadata() const {
     return _qp.process(
             query,
             db::consistency_level::QUORUM,
-            infinite_timeout_config).then([this](::shared_ptr<cql3::untyped_result_set> results) {
+            internal_distributed_timeout_config()).then([this](::shared_ptr<cql3::untyped_result_set> results) {
         return do_for_each(*results, [this](const cql3::untyped_result_set_row& row) {
             auto username = row.get_as<sstring>("username");
             auto salted_hash = row.get_as<sstring>(SALTED_HASH);
 
             return _qp.process(
                     update_row_query,
                     consistency_for_user(username),
-                    infinite_timeout_config,
+                    internal_distributed_timeout_config(),
                     {std::move(salted_hash), username}).discard_result();
         }).finally([results] {});
     }).then([] {
@@ -208,7 +208,7 @@ future<> password_authenticator::create_default_if_missing() const {
             return _qp.process(
                     update_row_query,
                     db::consistency_level::QUORUM,
-                    infinite_timeout_config,
+                    internal_distributed_timeout_config(),
                     {hashpw(DEFAULT_USER_PASSWORD), DEFAULT_USER_NAME}).then([](auto&&) {
                 plogger.info("Created default superuser authentication record.");
             });
@@ -308,7 +308,7 @@ future<authenticated_user> password_authenticator::authenticate(
         return _qp.process(
                 query,
                 consistency_for_user(username),
-                infinite_timeout_config,
+                internal_distributed_timeout_config(),
                 {username},
                 true);
     }).then_wrapped([=](future<::shared_ptr<cql3::untyped_result_set>> f) {
@@ -336,7 +336,7 @@ future<> password_authenticator::create(stdx::string_view role_name, const authe
     return _qp.process(
             update_row_query,
             consistency_for_user(role_name),
-            infinite_timeout_config,
+            internal_distributed_timeout_config(),
             {hashpw(*options.password), sstring(role_name)}).discard_result();
 }
 
@@ -354,7 +354,7 @@ future<> password_authenticator::alter(stdx::string_view role_name, const authen
     return _qp.process(
             query,
             consistency_for_user(role_name),
-            infinite_timeout_config,
+            internal_distributed_timeout_config(),
             {hashpw(*options.password), sstring(role_name)}).discard_result();
 }
 
@@ -365,7 +365,10 @@ future<> password_authenticator::drop(stdx::string_view name) const {
             meta::roles_table::qualified_name(),
             meta::roles_table::role_col_name);
 
-    return _qp.process(query, consistency_for_user(name), infinite_timeout_config, {sstring(name)}).discard_result();
+    return _qp.process(
+            query, consistency_for_user(name),
+            internal_distributed_timeout_config(),
+            {sstring(name)}).discard_result();
 }
 
 future<custom_options> password_authenticator::query_custom_options(stdx::string_view role_name) const {

auth/standard_role_manager.cc

@@ -89,7 +89,7 @@ static future<stdx::optional<record>> find_record(cql3::query_processor& qp, std
     return qp.process(
             query,
             consistency_for_role(role_name),
-            infinite_timeout_config,
+            internal_distributed_timeout_config(),
             {sstring(role_name)},
             true).then([](::shared_ptr<cql3::untyped_result_set> results) {
         if (results->empty()) {
@@ -174,7 +174,7 @@ future<> standard_role_manager::create_default_role_if_missing() const {
             return _qp.process(
                     query,
                     db::consistency_level::QUORUM,
-                    infinite_timeout_config,
+                    internal_distributed_timeout_config(),
                     {meta::DEFAULT_SUPERUSER_NAME}).then([](auto&&) {
                 log.info("Created default superuser role '{}'.", meta::DEFAULT_SUPERUSER_NAME);
                 return make_ready_future<>();
@@ -201,7 +201,7 @@ future<> standard_role_manager::migrate_legacy_metadata() const {
     return _qp.process(
             query,
             db::consistency_level::QUORUM,
-            infinite_timeout_config).then([this](::shared_ptr<cql3::untyped_result_set> results) {
+            internal_distributed_timeout_config()).then([this](::shared_ptr<cql3::untyped_result_set> results) {
         return do_for_each(*results, [this](const cql3::untyped_result_set_row& row) {
             role_config config;
             config.is_superuser = row.get_as<bool>("super");
@@ -263,7 +263,7 @@ future<> standard_role_manager::create_or_replace(stdx::string_view role_name, c
     return _qp.process(
             query,
             consistency_for_role(role_name),
-            infinite_timeout_config,
+            internal_distributed_timeout_config(),
             {sstring(role_name), c.is_superuser, c.can_login},
             true).discard_result();
 }
@@ -307,7 +307,7 @@ standard_role_manager::alter(stdx::string_view role_name, const role_config_upda
                         build_column_assignments(u),
                         meta::roles_table::role_col_name),
                 consistency_for_role(role_name),
-                infinite_timeout_config,
+                internal_distributed_timeout_config(),
                 {sstring(role_name)}).discard_result();
     });
 }
@@ -327,7 +327,7 @@ future<> standard_role_manager::drop(stdx::string_view role_name) const {
             return _qp.process(
                     query,
                     consistency_for_role(role_name),
-                    infinite_timeout_config,
+                    internal_distributed_timeout_config(),
                     {sstring(role_name)}).then([this, role_name](::shared_ptr<cql3::untyped_result_set> members) {
                 return parallel_for_each(
                         members->begin(),
@@ -367,7 +367,7 @@ future<> standard_role_manager::drop(stdx::string_view role_name) const {
             return _qp.process(
                     query,
                     consistency_for_role(role_name),
-                    infinite_timeout_config,
+                    internal_distributed_timeout_config(),
                     {sstring(role_name)}).discard_result();
         };
 
@@ -394,7 +394,7 @@ standard_role_manager::modify_membership(
         return _qp.process(
                 query,
                 consistency_for_role(grantee_name),
-                infinite_timeout_config,
+                internal_distributed_timeout_config(),
                 {role_set{sstring(role_name)}, sstring(grantee_name)}).discard_result();
     };
 
@@ -406,7 +406,7 @@ standard_role_manager::modify_membership(
                                 "INSERT INTO %s (role, member) VALUES (?, ?)",
                                 meta::role_members_table::qualified_name()),
                         consistency_for_role(role_name),
-                        infinite_timeout_config,
+                        internal_distributed_timeout_config(),
                         {sstring(role_name), sstring(grantee_name)}).discard_result();
 
             case membership_change::remove:
@@ -415,7 +415,7 @@ standard_role_manager::modify_membership(
                                 "DELETE FROM %s WHERE role = ? AND member = ?",
                                 meta::role_members_table::qualified_name()),
                         consistency_for_role(role_name),
-                        infinite_timeout_config,
+                        internal_distributed_timeout_config(),
                         {sstring(role_name), sstring(grantee_name)}).discard_result();
         }
 
@@ -516,7 +516,10 @@ future<role_set> standard_role_manager::query_all() const {
     // To avoid many copies of a view.
     static const auto role_col_name_string = sstring(meta::roles_table::role_col_name);
 
-    return _qp.process(query, db::consistency_level::QUORUM, infinite_timeout_config).then([](::shared_ptr<cql3::untyped_result_set> results) {
+    return _qp.process(
+            query,
+            db::consistency_level::QUORUM,
+            internal_distributed_timeout_config()).then([](::shared_ptr<cql3::untyped_result_set> results) {
         role_set roles;
 
         std::transform(

db/system_distributed_keyspace.cc

@@ -26,6 +26,7 @@
 #include "db/consistency_level_type.hh"
 #include "db/system_keyspace.hh"
 #include "schema_builder.hh"
+#include "timeout_config.hh"
 #include "types.hh"
 
 #include <seastar/core/reactor.hh>
@@ -97,11 +98,17 @@ future<> system_distributed_keyspace::stop() {
     return make_ready_future<>();
 }
 
+static const timeout_config internal_distributed_timeout_config = [] {
+    using namespace std::chrono_literals;
+    const auto t = 10s;
+    return timeout_config{ t, t, t, t, t, t, t };
+}();
+
 future<std::unordered_map<utils::UUID, sstring>> system_distributed_keyspace::view_status(sstring ks_name, sstring view_name) const {
     return _qp.process(
             sprint("SELECT host_id, status FROM %s.%s WHERE keyspace_name = ? AND view_name = ?", NAME, VIEW_BUILD_STATUS),
             db::consistency_level::ONE,
-            infinite_timeout_config,
+            internal_distributed_timeout_config,
             { std::move(ks_name), std::move(view_name) },
             false).then([this] (::shared_ptr<cql3::untyped_result_set> cql_result) {
         return boost::copy_range<std::unordered_map<utils::UUID, sstring>>(*cql_result
@@ -118,7 +125,7 @@ future<> system_distributed_keyspace::start_view_build(sstring ks_name, sstring
         return _qp.process(
                 sprint("INSERT INTO %s.%s (keyspace_name, view_name, host_id, status) VALUES (?, ?, ?, ?)", NAME, VIEW_BUILD_STATUS),
                 db::consistency_level::ONE,
-                infinite_timeout_config,
+                internal_distributed_timeout_config,
                 { std::move(ks_name), std::move(view_name), std::move(host_id), "STARTED" },
                 false).discard_result();
     });
@@ -129,7 +136,7 @@ future<> system_distributed_keyspace::finish_view_build(sstring ks_name, sstring
         return _qp.process(
                 sprint("UPDATE %s.%s SET status = ? WHERE keyspace_name = ? AND view_name = ? AND host_id = ?", NAME, VIEW_BUILD_STATUS),
                 db::consistency_level::ONE,
-                infinite_timeout_config,
+                internal_distributed_timeout_config,
                 { "SUCCESS", std::move(ks_name), std::move(view_name), std::move(host_id) },
                 false).discard_result();
     });
@@ -139,7 +146,7 @@ future<> system_distributed_keyspace::remove_view(sstring ks_name, sstring view_
     return _qp.process(
             sprint("DELETE FROM %s.%s WHERE keyspace_name = ? AND view_name = ?", NAME, VIEW_BUILD_STATUS),
             db::consistency_level::ONE,
-            infinite_timeout_config,
+            internal_distributed_timeout_config,
             { std::move(ks_name), std::move(view_name) },
             false).discard_result();
 }