fix: Proper shutdown sequence with Namespaces (#3333)

This removes a race between periodic fiber and namespaces during
shutdown.

src/server/blocking_controller_test.cc

@@ -70,8 +70,8 @@ void BlockingControllerTest::SetUp() {
 }
 
 void BlockingControllerTest::TearDown() {
+  shard_set->PreShutdown();
   namespaces.Clear();
-
   shard_set->Shutdown();
   delete shard_set;
 

src/server/engine_shard_set.cc

@@ -396,12 +396,16 @@ void EngineShard::Shutdown() {
     tiered_storage_.reset();
   }
 
+  DCHECK(!fiber_periodic_.IsJoinable());
+
+  ProactorBase::me()->RemoveOnIdleTask(defrag_task_);
+}
+
+void EngineShard::StopPeriodicFiber() {
   fiber_periodic_done_.Notify();
   if (fiber_periodic_.IsJoinable()) {
     fiber_periodic_.Join();
   }
-
-  ProactorBase::me()->RemoveOnIdleTask(defrag_task_);
 }
 
 void EngineShard::StartPeriodicFiber(util::ProactorBase* pb) {
@@ -895,6 +899,10 @@ void EngineShardSet::Init(uint32_t sz, bool update_db_time) {
   });
 }
 
+void EngineShardSet::PreShutdown() {
+  RunBlockingInParallel([](EngineShard* shard) { shard->StopPeriodicFiber(); });
+}
+
 void EngineShardSet::Shutdown() {
   RunBlockingInParallel([](EngineShard*) { EngineShard::DestroyThreadLocal(); });
 }

src/server/engine_shard_set.h

@@ -149,6 +149,8 @@ class EngineShard {
 
   void TEST_EnableHeartbeat();
 
+  void StopPeriodicFiber();
+
   struct TxQueueInfo {
     // Armed - those that the coordinator has armed with callbacks and wants them to run.
     // Runnable - those that could run (they own the locks) but probably can not run due
@@ -281,6 +283,12 @@ class EngineShardSet {
   }
 
   void Init(uint32_t size, bool update_db_time);
+
+  // Shutdown sequence:
+  // - EngineShardSet.PreShutDown()
+  // - Namespaces.Clear()
+  // - EngineShardSet.Shutdown()
+  void PreShutdown();
   void Shutdown();
 
   static const std::vector<CachedStats>& GetCachedStats();

src/server/main_service.cc

@@ -833,7 +833,6 @@ Service::Service(ProactorPool* pp)
 Service::~Service() {
   delete shard_set;
   shard_set = nullptr;
-  namespaces.Clear();
 }
 
 void Service::Init(util::AcceptServer* acceptor, std::vector<facade::Listener*> listeners,
@@ -913,8 +912,8 @@ void Service::Shutdown() {
 
   ChannelStore::Destroy();
 
+  shard_set->PreShutdown();
   namespaces.Clear();
-
   shard_set->Shutdown();
 
   pp_.Await([](ProactorBase* pb) { ServerState::tlocal()->Destroy(); });