api, database, storage_service: Unify auto-compaction toggle

There are two knobs here -- global and per-table one. Both were added
without any synchronisation, but the former one was later fixed to
become serialized and not to be available "too early".

This patch unifies both toggles to be serialized with each-other and
not be enabled too early.

The justification for this change is to move the global toggle from out
of the storage service, as it really belongs to the database, not the
storage service. Respectively, the current synchronization, that depends
on storage service internals, should be replaced with something else.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>

api/column_family.cc

@@ -862,19 +862,25 @@ void set_column_family(http_context& ctx, routes& r) {
     });
 
     cf::enable_auto_compaction.set(r, [&ctx](std::unique_ptr<request> req) {
+      return ctx.db.invoke_on(0, [&ctx, req = std::move(req)] (database& db) {
+        auto g = database::autocompaction_toggle_guard(db);
         return foreach_column_family(ctx, req->param["name"], [](column_family &cf) {
             cf.enable_auto_compaction();
-        }).then([] {
+        }).then([g = std::move(g)] {
             return make_ready_future<json::json_return_type>(json_void());
         });
+      });
     });
 
     cf::disable_auto_compaction.set(r, [&ctx](std::unique_ptr<request> req) {
+      return ctx.db.invoke_on(0, [&ctx, req = std::move(req)] (database& db) {
+        auto g = database::autocompaction_toggle_guard(db);
         return foreach_column_family(ctx, req->param["name"], [](column_family &cf) {
             cf.disable_auto_compaction();
-        }).then([] {
+        }).then([g = std::move(g)] {
             return make_ready_future<json::json_return_type>(json_void());
         });
+      });
     });
 
     cf::get_built_indexes.set(r, [&ctx](std::unique_ptr<request> req) {

database.hh

@@ -1325,6 +1325,7 @@ private:
     seastar::metrics::metric_groups _metrics;
     bool _enable_incremental_backups = false;
     bool _shutdown = false;
+    bool _enable_autocompaction_toggle = false;
     query::querier_cache _querier_cache;
 
     std::unique_ptr<db::large_data_handler> _large_data_handler;
@@ -1388,6 +1389,25 @@ public:
 
     void set_enable_incremental_backups(bool val) { _enable_incremental_backups = val; }
 
+    void enable_autocompaction_toggle() noexcept { _enable_autocompaction_toggle = true; }
+    class autocompaction_toggle_guard {
+        database& _db;
+    public:
+        autocompaction_toggle_guard(database& db) : _db(db) {
+            assert(this_shard_id() == 0);
+            if (!_db._enable_autocompaction_toggle) {
+                throw std::runtime_error("Autocompaction toggle is busy");
+            }
+            _db._enable_autocompaction_toggle = false;
+        }
+        autocompaction_toggle_guard(const autocompaction_toggle_guard&) = delete;
+        autocompaction_toggle_guard(autocompaction_toggle_guard&&) = default;
+        ~autocompaction_toggle_guard() {
+            assert(this_shard_id() == 0);
+            _db._enable_autocompaction_toggle = true;
+        }
+    };
+
     future<> parse_system_tables(distributed<service::storage_proxy>&);
     database(const db::config&, database_config dbcfg, service::migration_notifier& mn, gms::feature_service& feat, const locator::shared_token_metadata& stm,
             abort_source& as, sharded<semaphore>& sst_dir_sem, utils::cross_shard_barrier barrier = utils::cross_shard_barrier(utils::cross_shard_barrier::solo{}) /* for single-shard usage */);

main.cc

@@ -1164,6 +1164,15 @@ int main(int ac, char** av) {
                 mm_notifier.local().unregister_listener(&ss.local()).get();
             });
 
+            /*
+             * FIXME. In bb07678346 commit the API toggle for autocompaction was
+             * (partially) delayed until system prepared to join the ring. Probably
+             * it was an overkill and it can be enabled earlier, even as early as
+             * 'by default'. E.g. the per-table toggle was 'enabled' right after
+             * the system keyspace started and nobody seemed to have any troubles.
+             */
+            db.local().enable_autocompaction_toggle();
+
             with_scheduling_group(maintenance_scheduling_group, [&] {
                 return messaging.invoke_on_all(&netw::messaging_service::start_listen);
             }).get();

service/storage_service.cc

@@ -3051,12 +3051,9 @@ future<>
 storage_service::set_tables_autocompaction(const sstring &keyspace, std::vector<sstring> tables, bool enabled) {
     slogger.info("set_tables_autocompaction: enabled={} keyspace={} tables={}", enabled, keyspace, tables);
     return do_with(keyspace, std::move(tables), [this, enabled] (const sstring &keyspace, const std::vector<sstring>& tables) {
-        return run_with_api_lock(sstring("set_tables_autocompaction"), [&keyspace, &tables, enabled] (auto&& ss) {
-            if (!ss._initialized) {
-                return make_exception_future<>(std::runtime_error("Too early: storage service not initialized yet"));
-            }
-
-            return ss._db.invoke_on_all([&keyspace, &tables, enabled] (database& db) {
+        return _db.invoke_on(0, [this, &keyspace, &tables, enabled] (database& db) {
+            auto g = database::autocompaction_toggle_guard(db);
+            return _db.invoke_on_all([&keyspace, &tables, enabled] (database& db) {
                 return parallel_for_each(tables, [&db, &keyspace, enabled] (const sstring& table) {
                     column_family& cf = db.find_column_family(keyspace, table);
                     if (enabled) {
@@ -3066,7 +3063,7 @@ storage_service::set_tables_autocompaction(const sstring &keyspace, std::vector<
                     }
                     return make_ready_future<>();
                 });
-            });
+            }).finally([g = std::move(g)] {});
         });
     });
 }