Merge 'virtual_tables, schema_registry: fix use after free related to schema registry' from Avi Kivity

Both virtual tables and schema registry contain thread_local caches that are destroyed
at thread exit. after a Seastar change[1], these destructions can happen after the reactor
is destroyed, triggering a use-after-free.

Fix by scoping the destruction so it takes place earlier.

[1] 101b245ed7

Closes scylladb/scylladb#16510

* github.com:scylladb/scylladb:
  schema_registry, database: flush entries when no longer in use
  virtual_tables: scope virtual tables registry in system_keyspace

db/system_keyspace.hh

@@ -24,6 +24,7 @@
 #include "cdc/generation_id.hh"
 #include "locator/host_id.hh"
 #include "mutation/canonical_mutation.hh"
+#include "virtual_tables.hh"
 
 namespace sstables {
     struct entry_descriptor;
@@ -112,6 +113,7 @@ class system_keyspace : public seastar::peering_sharded_service<system_keyspace>
     cql3::query_processor& _qp;
     replica::database& _db;
     std::unique_ptr<local_cache> _cache;
+    virtual_tables_registry _virtual_tables_registry;
 
     static schema_ptr raft_snapshot_config();
     static schema_ptr local();
@@ -521,6 +523,7 @@ public:
     ~system_keyspace();
     future<> shutdown();
 
+    virtual_tables_registry& get_virtual_tables_registry() { return _virtual_tables_registry; }
 private:
     future<::shared_ptr<cql3::untyped_result_set>> execute_cql(const sstring& query_string, const std::initializer_list<data_value>& values);
     template <typename... Args>

db/virtual_tables.cc

@@ -982,11 +982,8 @@ private:
 
 }
 
-// Map from table's schema ID to table itself. Helps avoiding accidental duplication.
-static thread_local std::map<table_id, std::unique_ptr<virtual_table>> virtual_tables;
-
-static void register_virtual_tables(distributed<replica::database>& dist_db, distributed<service::storage_service>& dist_ss, sharded<gms::gossiper>& dist_gossiper, sharded<service::raft_group_registry>& dist_raft_gr, db::config& cfg) {
-    auto add_table = [] (std::unique_ptr<virtual_table>&& tbl) {
+static void register_virtual_tables(virtual_tables_registry_impl& virtual_tables, distributed<replica::database>& dist_db, distributed<service::storage_service>& dist_ss, sharded<gms::gossiper>& dist_gossiper, sharded<service::raft_group_registry>& dist_raft_gr, db::config& cfg) {
+    auto add_table = [&] (std::unique_ptr<virtual_table>&& tbl) {
         virtual_tables[tbl->schema()->id()] = std::move(tbl);
     };
 
@@ -1007,6 +1004,7 @@ static void register_virtual_tables(distributed<replica::database>& dist_db, dis
 }
 
 static void install_virtual_readers_and_writers(db::system_keyspace& sys_ks, replica::database& db) {
+    auto& virtual_tables = *sys_ks.get_virtual_tables_registry();
     db.find_column_family(system_keyspace::size_estimates()).set_virtual_reader(mutation_source(db::size_estimates::virtual_reader(db, sys_ks)));
     db.find_column_family(system_keyspace::v3::views_builds_in_progress()).set_virtual_reader(mutation_source(db::view::build_progress_virtual_reader(db)));
     db.find_column_family(system_keyspace::built_indexes()).set_virtual_reader(mutation_source(db::index::built_indexes_virtual_reader(db)));
@@ -1023,7 +1021,9 @@ future<> initialize_virtual_tables(
         sharded<gms::gossiper>& dist_gossiper, distributed<service::raft_group_registry>& dist_raft_gr,
         sharded<db::system_keyspace>& sys_ks,
         db::config& cfg) {
-    register_virtual_tables(dist_db, dist_ss, dist_gossiper, dist_raft_gr, cfg);
+    auto& virtual_tables_registry = sys_ks.local().get_virtual_tables_registry();
+    auto& virtual_tables = *virtual_tables_registry;
+    register_virtual_tables(virtual_tables, dist_db, dist_ss, dist_gossiper, dist_raft_gr, cfg);
 
     auto& db = dist_db.local();
     for (auto&& [id, vt] : virtual_tables) {
@@ -1034,4 +1034,9 @@ future<> initialize_virtual_tables(
     install_virtual_readers_and_writers(sys_ks.local(), db);
 }
 
+virtual_tables_registry::virtual_tables_registry() : unique_ptr(std::make_unique<virtual_tables_registry_impl>()) {
+}
+
+virtual_tables_registry::~virtual_tables_registry() = default;
+
 } // namespace db

db/virtual_tables.hh

@@ -10,6 +10,7 @@
 #pragma once
 
 #include <seastar/core/distributed.hh>
+#include <map>
 #include "schema/schema_fwd.hh"
 
 namespace replica {
@@ -38,4 +39,16 @@ future<> initialize_virtual_tables(
     sharded<db::system_keyspace>& sys_ks,
     db::config&);
 
+
+class virtual_table;
+
+using virtual_tables_registry_impl = std::map<table_id, std::unique_ptr<virtual_table>>;
+
+// Pimpl to hide virtual_table from the rest of the code
+class virtual_tables_registry : public std::unique_ptr<virtual_tables_registry_impl> {
+public:
+    virtual_tables_registry();
+    ~virtual_tables_registry();
+};
+
 } // namespace db

replica/database.cc

@@ -636,6 +636,7 @@ void database::set_format(sstables::sstable_version_types format) noexcept {
 
 database::~database() {
     _user_types->deactivate();
+    local_schema_registry().clear();
 }
 
 void database::update_version(const table_schema_version& version) {

schema/schema_registry.cc

@@ -164,6 +164,10 @@ schema_ptr schema_registry::get_or_load(table_schema_version v, const schema_loa
     return e.get_schema();
 }
 
+void schema_registry::clear() {
+    _entries.clear();
+}
+
 schema_ptr schema_registry_entry::load(frozen_schema fs) {
     _frozen_schema = std::move(fs);
     auto s = get_schema();

schema/schema_registry.hh

@@ -150,6 +150,12 @@ public:
     // The schema instance pointed to by the argument will be attached to the registry
     // entry and will keep it alive.
     schema_ptr learn(const schema_ptr&);
+
+    // Removes all entries from the registry. This in turn removes all dependencies
+    // on the Seastar reactor.
+    //
+    // Prerequisite: all futures from get_or_load() are resolved.
+    void clear();
 };
 
 schema_registry& local_schema_registry();