service/storage_service: drain view_building_worker earlier

Similarly to view builder, view building worker needs to be drained
in `storage_service::do_drain()`.

Storage service drain is happening at the same beginning of shutdown
procedure. Before this patch, the worker was still building views
after the storage service was drained and this caused errors like:
`Error applying view update to (named_gate_closed_exception)` and
`locator::no_such_tablet_map`.

Fixes scylladb/scylladb#25908

Closes scylladb/scylladb#25984

main.cc

@@ -1819,7 +1819,7 @@ sharded<locator::shared_token_metadata> token_metadata;
                 std::ref(feature_service), std::ref(mm), std::ref(token_metadata), std::ref(erm_factory),
                 std::ref(messaging), std::ref(repair),
                 std::ref(stream_manager), std::ref(lifecycle_notifier), std::ref(bm), std::ref(snitch),
-                std::ref(tablet_allocator), std::ref(cdc_generation_service), std::ref(view_builder), std::ref(qp), std::ref(sl_controller),
+                std::ref(tablet_allocator), std::ref(cdc_generation_service), std::ref(view_builder), std::ref(view_building_worker), std::ref(qp), std::ref(sl_controller),
                 std::ref(tsm), std::ref(vbsm), std::ref(task_manager), std::ref(gossip_address_map),
                 compression_dict_updated_callback,
                 only_on_shard0(&*disk_space_monitor_shard0)

service/storage_service.cc

@@ -10,6 +10,7 @@
  */
 
 #include "storage_service.hh"
+#include "db/view/view_building_worker.hh"
 #include "utils/chunked_vector.hh"
 #include <seastar/core/shard_id.hh>
 #include "db/view/view_building_coordinator.hh"
@@ -197,6 +198,7 @@ storage_service::storage_service(abort_source& abort_source,
     sharded<service::tablet_allocator>& tablet_allocator,
     sharded<cdc::generation_service>& cdc_gens,
     sharded<db::view::view_builder>& view_builder,
+    sharded<db::view::view_building_worker>& view_building_worker,
     cql3::query_processor& qp,
     sharded<qos::service_level_controller>& sl_controller,
     topology_state_machine& topology_state_machine,
@@ -237,6 +239,7 @@ storage_service::storage_service(abort_source& abort_source,
         , _tablet_allocator(tablet_allocator)
         , _cdc_gens(cdc_gens)
         , _view_builder(view_builder)
+        , _view_building_worker(view_building_worker)
         , _topology_state_machine(topology_state_machine)
         , _view_building_state_machine(view_building_state_machine)
         , _compression_dictionary_updated_callback(std::move(compression_dictionary_updated_callback))
@@ -4754,6 +4757,7 @@ future<> storage_service::do_drain() {
     // Drain view builder before group0, because the view builder uses group0 to coordinate view building.
     // Drain after transport is stopped, because view_builder::drain aborts view writes for user writes as well.
     co_await _view_builder.invoke_on_all(&db::view::view_builder::drain);
+    co_await _view_building_worker.invoke_on_all(&db::view::view_building_worker::drain);
 
     // group0 persistence relies on local storage, so we need to stop group0 first.
     // This must be kept in sync with defer_verbose_shutdown for group0 in main.cc to

service/storage_service.hh

@@ -77,6 +77,7 @@ class system_keyspace;
 class batchlog_manager;
 namespace view {
 class view_builder;
+class view_building_worker;
 }
 }
 
@@ -240,6 +241,7 @@ public:
         sharded<service::tablet_allocator>& tablet_allocator,
         sharded<cdc::generation_service>& cdc_gs,
         sharded<db::view::view_builder>& view_builder,
+        sharded<db::view::view_building_worker>& view_building_worker,
         cql3::query_processor& qp,
         sharded<qos::service_level_controller>& sl_controller,
         topology_state_machine& topology_state_machine,
@@ -581,6 +583,7 @@ private:
     sharded<service::tablet_allocator>& _tablet_allocator;
     sharded<cdc::generation_service>& _cdc_gens;
     sharded<db::view::view_builder>& _view_builder;
+    sharded<db::view::view_building_worker>& _view_building_worker;
     bool _isolated = false;
 private:
     /**

test/lib/cql_test_env.cc

@@ -948,7 +948,7 @@ private:
                 std::ref(_snitch),
                 std::ref(_tablet_allocator),
                 std::ref(_cdc_generation_service),
-                std::ref(_view_builder),
+                std::ref(_view_builder), std::ref(_view_building_worker),
                 std::ref(_qp),
                 std::ref(_sl_controller),
                 std::ref(_topology_state_machine),