Fix online SSTable loading with concurrent tablet migration

load-and-stream is currently the only method -- for tablets -- that
can load SSTables while the node is online.
Today, sstable_directory relies on replication map (erm) not being
invalidated during loading, and the assumption is broken with
concurrent tablet migration.
It causes load-and-stream to segfault.

The sstable loader needs the sharder from erm in order to compute
the owning shard.

To fix, let's use auto_refreshing_sharder, which refreshes sharder
every time table has replication map updated. So we guarantee any
user of sharder will find it alive throughout the lifetime of
sstable_directory.

Signed-off-by: Raphael S. Carvalho <raphaelsc@scylladb.com>

sstables/sstable_directory.cc

@@ -23,6 +23,7 @@
 #include "utils/directories.hh"
 #include "replica/database.hh"
 #include "db/system_keyspace.hh"
+#include "dht/auto_refreshing_sharder.hh"
 
 static logging::logger dirlog("sstable_directory");
 
@@ -67,7 +68,7 @@ sstable_directory::sstable_directory(replica::table& table,
     : sstable_directory(
         table.get_sstables_manager(),
         table.schema(),
-        table.get_effective_replication_map()->get_sharder(*table.schema()),
+        std::make_unique<dht::auto_refreshing_sharder>(table.shared_from_this()),
         table.get_storage_options_ptr(),
         table.dir(),
         std::move(state),
@@ -82,6 +83,26 @@ sstable_directory::sstable_directory(sstables_manager& manager,
         sstring table_dir,
         sstable_state state,
         io_error_handler_gen error_handler_gen)
+    : sstable_directory(
+        manager,
+        std::move(schema),
+        &sharder,
+        std::move(storage_opts),
+        std::move(table_dir),
+        state,
+        std::move(error_handler_gen)
+    )
+{}
+
+using unique_sharder_ptr = std::unique_ptr<dht::sharder>;
+
+sstable_directory::sstable_directory(sstables_manager& manager,
+        schema_ptr schema,
+        std::variant<unique_sharder_ptr, const dht::sharder*> sharder,
+        lw_shared_ptr<const data_dictionary::storage_options> storage_opts,
+        sstring table_dir,
+        sstable_state state,
+        io_error_handler_gen error_handler_gen)
     : _manager(manager)
     , _schema(std::move(schema))
     , _storage_opts(std::move(storage_opts))
@@ -91,7 +112,8 @@ sstable_directory::sstable_directory(sstables_manager& manager,
     , _error_handler_gen(error_handler_gen)
     , _storage(make_storage(_manager, *_storage_opts, _table_dir, _state))
     , _lister(make_components_lister())
-    , _sharder(sharder)
+    , _sharder_ptr(std::holds_alternative<unique_sharder_ptr>(sharder) ? std::move(std::get<unique_sharder_ptr>(sharder)) : nullptr)
+    , _sharder(_sharder_ptr ? *_sharder_ptr : *std::get<const dht::sharder*>(sharder))
     , _unshared_remote_sstables(smp::count)
 {}
 

sstables/sstable_directory.hh

@@ -142,6 +142,7 @@ private:
     io_error_handler_gen _error_handler_gen;
     std::unique_ptr<storage> _storage;
     std::unique_ptr<components_lister> _lister;
+    std::unique_ptr<dht::sharder> _sharder_ptr;
     const dht::sharder& _sharder;
 
     generation_type _max_generation_seen;
@@ -189,6 +190,13 @@ private:
     // Retrieves sstables::foreign_sstable_open_info for a particular SSTable.
     future<foreign_sstable_open_info> get_open_info_for_this_sstable(const sstables::entry_descriptor& desc) const;
 
+    sstable_directory(sstables_manager& manager,
+          schema_ptr schema,
+          std::variant<std::unique_ptr<dht::sharder>, const dht::sharder*> sharder,
+          lw_shared_ptr<const data_dictionary::storage_options> storage_opts,
+          sstring table_dir,
+          sstable_state state,
+          io_error_handler_gen error_handler_gen);
 public:
     sstable_directory(replica::table& table,
             sstable_state state,