db/schema_tables: Include view_virtual_columns in the digest only when all nodes do

After 7c87405, schema sync includes system_schema.view_virtual_columns
in the schema digest. Old nodes don't know about this table and will
not include it in the digest calculation. As a result, there will be
schema disagreement until the whole cluster is upgraded.

Fix this by taking the new table into account only when the whole
cluster is upgraded.

The table should not be used for anything before this happens. This is
not currently enforced, but should be.

Fixes #4457.

db/schema_tables.cc

@@ -2708,10 +2708,17 @@ std::vector<schema_ptr> all_tables(schema_features features) {
     // Don't forget to update this list when new schema tables are added.
     // The listed schema tables are the ones synchronized between nodes,
     // and forgetting one of them in this list can cause bugs like #4339.
-    return {
+    //
+    // This list must be kept backwards-compatible because it's used
+    // for schema digest calculation. Refs #4457.
+    std::vector<schema_ptr> result = {
         keyspaces(), tables(), scylla_tables(), columns(), dropped_columns(), triggers(),
-        views(), types(), functions(), aggregates(), indexes(), view_virtual_columns()
+        views(), types(), functions(), aggregates(), indexes()
     };
+    if (features.contains<schema_feature::VIEW_VIRTUAL_COLUMNS>()) {
+        result.emplace_back(view_virtual_columns());
+    }
+    return result;
 }
 
 std::vector<sstring> all_table_names(schema_features features) {

service/migration_manager.cc

@@ -74,11 +74,21 @@ future<> migration_manager::stop()
     return parallel_for_each(_schema_pulls.begin(), _schema_pulls.end(), [] (auto&& e) {
         serialized_action& sp = e.second;
         return sp.join();
+    }).finally([this] {
+        return _background_tasks.close();
     });
 }
 
 void migration_manager::init_messaging_service()
 {
+    auto& ss = service::get_local_storage_service();
+    _feature_listeners.push_back(ss.cluster_supports_view_virtual_columns().when_enabled([this, &ss] {
+        with_gate(_background_tasks, [this, &ss] {
+            mlogger.debug("view_virtual_columns feature enabled, recalculating schema version");
+            return update_schema_version(get_storage_proxy(), ss.cluster_schema_features());
+        });
+    }));
+
     auto& ms = netw::get_local_messaging_service();
     ms.register_definitions_update([this] (const rpc::client_info& cinfo, std::vector<frozen_mutation> m) {
         auto src = netw::messaging_service::get_source(cinfo);

service/migration_manager.hh

@@ -57,6 +57,8 @@ namespace service {
 class migration_manager : public seastar::async_sharded_service<migration_manager> {
     std::vector<migration_listener*> _listeners;
     std::unordered_map<netw::msg_addr, serialized_action, netw::msg_addr::hash> _schema_pulls;
+    std::vector<gms::feature::listener_registration> _feature_listeners;
+    seastar::gate _background_tasks;
     static const std::chrono::milliseconds migration_delay;
 public:
     migration_manager();