auth: Remove ordering dependence

If `auth::password_authenticator` also creates `system_auth.roles` and
we fix the existence check for the default superuser in
`auth::standard_role_manager` to only search for the columns that it
owns (instead of the column itself), then both modules' initialization
are independent of one another.

Fixes #3319.

auth/common.cc

@@ -59,13 +59,13 @@ future<> do_after_system_ready(seastar::abort_source& as, seastar::noncopyable_f
 }
 
 future<> create_metadata_table_if_missing(
-        const sstring& table_name,
+        stdx::string_view table_name,
         cql3::query_processor& qp,
-        const sstring& cql,
+        stdx::string_view cql,
         ::service::migration_manager& mm) {
     auto& db = qp.db().local();
 
-    if (db.has_schema(meta::AUTH_KS, table_name)) {
+    if (db.has_schema(meta::AUTH_KS, sstring(table_name))) {
         return make_ready_future<>();
     }
 

auth/common.hh

@@ -22,6 +22,7 @@
 #pragma once
 
 #include <chrono>
+#include <experimental/string_view>
 
 #include <seastar/core/future.hh>
 #include <seastar/core/abort_source.hh>
@@ -74,9 +75,9 @@ inline future<> delay_until_system_ready(seastar::abort_source& as) {
 future<> do_after_system_ready(seastar::abort_source& as, seastar::noncopyable_function<future<>()> func);
 
 future<> create_metadata_table_if_missing(
-        const sstring& table_name,
+        stdx::string_view table_name,
         cql3::query_processor&,
-        const sstring& cql,
+        stdx::string_view cql,
         ::service::migration_manager&);
 
 future<> wait_for_schema_agreement(::service::migration_manager&, const database&);

auth/password_authenticator.cc

@@ -220,6 +220,12 @@ future<> password_authenticator::start() {
      return once_among_shards([this] {
          gensalt(); // do this once to determine usable hashing
 
+         auto f = create_metadata_table_if_missing(
+                 meta::roles_table::name,
+                 _qp,
+                 meta::roles_table::creation_query(),
+                 _migration_manager);
+
          _stopped = do_after_system_ready(_as, [this] {
              return async([this] {
                  wait_for_schema_agreement(_migration_manager, _qp.db().local()).get0();
@@ -241,7 +247,7 @@ future<> password_authenticator::start() {
              });
          });
 
-         return make_ready_future<>();
+         return f;
      });
  }
 

auth/roles-metadata.cc

@@ -36,6 +36,21 @@ namespace meta {
 
 namespace roles_table {
 
+stdx::string_view creation_query() {
+    static const sstring instance = sprint(
+            "CREATE TABLE %s ("
+            "  %s text PRIMARY KEY,"
+            "  can_login boolean,"
+            "  is_superuser boolean,"
+            "  member_of set<text>,"
+            "  salted_hash text"
+            ")",
+            qualified_name(),
+            role_col_name);
+
+    return instance;
+}
+
 stdx::string_view qualified_name() noexcept {
     static const sstring instance = AUTH_KS + "." + sstring(name);
     return instance;

auth/roles-metadata.hh

@@ -40,6 +40,8 @@ namespace meta {
 
 namespace roles_table {
 
+stdx::string_view creation_query();
+
 constexpr stdx::string_view name{"roles", 5};
 
 stdx::string_view qualified_name() noexcept;

auth/service.cc

@@ -184,9 +184,7 @@ future<> service::start() {
     return once_among_shards([this] {
         return create_keyspace_if_missing();
     }).then([this] {
-        return _role_manager->start();
-    }).then([this] {
-        return when_all_succeed(_authorizer->start(), _authenticator->start());
+        return when_all_succeed(_role_manager->start(), _authorizer->start(), _authenticator->start());
     }).then([this] {
         _permissions_cache = std::make_unique<permissions_cache>(_permissions_cache_config, *this, log);
     }).then([this] {

auth/standard_role_manager.cc

@@ -118,6 +118,10 @@ static future<record> require_record(cql3::query_processor& qp, stdx::string_vie
    });
 }
 
+static bool has_can_login(const cql3::untyped_result_set_row& row) {
+    return row.has("can_login") && !(boolean_type->deserialize(row.get_blob("can_login")).is_null());
+}
+
 stdx::string_view standard_role_manager_name() noexcept {
     static const sstring instance = meta::AUTH_PACKAGE_NAME + "CassandraRoleManager";
     return instance;
@@ -136,17 +140,6 @@ const resource_set& standard_role_manager::protected_resources() const {
 }
 
 future<> standard_role_manager::create_metadata_tables_if_missing() const {
-    static const sstring create_roles_query = sprint(
-            "CREATE TABLE %s ("
-            "  %s text PRIMARY KEY,"
-            "  can_login boolean,"
-            "  is_superuser boolean,"
-            "  member_of set<text>,"
-            "  salted_hash text"
-            ")",
-            meta::roles_table::qualified_name(),
-            meta::roles_table::role_col_name);
-
     static const sstring create_role_members_query = sprint(
             "CREATE TABLE %s ("
             "  role text,"
@@ -158,19 +151,19 @@ future<> standard_role_manager::create_metadata_tables_if_missing() const {
 
     return when_all_succeed(
             create_metadata_table_if_missing(
-                    sstring(meta::roles_table::name),
+                    meta::roles_table::name,
                     _qp,
-                    create_roles_query,
+                    meta::roles_table::creation_query(),
                     _migration_manager),
             create_metadata_table_if_missing(
-                    sstring(meta::role_members_table::name),
+                    meta::role_members_table::name,
                     _qp,
                     create_role_members_query,
                     _migration_manager));
 }
 
 future<> standard_role_manager::create_default_role_if_missing() const {
-    return default_role_row_satisfies(_qp, [](auto&&) { return true; }).then([this](bool exists) {
+    return default_role_row_satisfies(_qp, &has_can_login).then([this](bool exists) {
         if (!exists) {
             static const sstring query = sprint(
                     "INSERT INTO %s (%s, is_superuser, can_login) VALUES (?, true, true)",
@@ -233,7 +226,7 @@ future<> standard_role_manager::start() {
                 return seastar::async([this] {
                     wait_for_schema_agreement(_migration_manager, _qp.db().local()).get0();
 
-                    if (any_nondefault_role_row_satisfies(_qp, [](auto&&) { return true; }).get0()) {
+                    if (any_nondefault_role_row_satisfies(_qp, &has_can_login).get0()) {
                         if (this->legacy_metadata_exists()) {
                             log.warn("Ignoring legacy user metadata since nondefault roles already exist.");
                         }