Merge pull request ClickHouse#87751 from ClickHouse/backport/25.8/78873

Backport ClickHouse#78873 to 25.8: Fix RMV tryEnqueueReplicatedDDL shutdown crash again

Author: clickhouse-gh[bot]

programs/server/Server.cpp

@@ -2820,11 +2820,11 @@ try
             else
                 LOG_INFO(log, "Closed connections.");
 
-            global_context->getRefreshSet().joinBackgroundTasks(wait_start + std::chrono::milliseconds(wait_limit_seconds * 1000));
+            bool joined_refresh_tasks = global_context->getRefreshSet().joinBackgroundTasks(wait_start + std::chrono::milliseconds(wait_limit_seconds * 1000));
 
             dns_cache_updater.reset();
 
-            if (current_connections)
+            if (current_connections || !joined_refresh_tasks)
             {
                 /// There is no better way to force connections to close in Poco.
                 /// Otherwise connection handlers will continue to live

src/Storages/MaterializedView/RefreshSet.cpp

@@ -203,7 +203,7 @@ bool RefreshSet::refreshesStopped() const
     return refreshes_stopped.load();
 }
 
-void RefreshSet::joinBackgroundTasks(std::chrono::steady_clock::time_point deadline)
+bool RefreshSet::joinBackgroundTasks(std::chrono::steady_clock::time_point deadline)
 {
     std::vector<RefreshTaskPtr> remaining_tasks;
     std::chrono::steady_clock::time_point stopped_at;
@@ -234,7 +234,11 @@ void RefreshSet::joinBackgroundTasks(std::chrono::steady_clock::time_point deadl
             names += remaining_tasks[i]->getInfo().view_id.getNameForLogs();
         }
         LOG_ERROR(getLogger("RefreshSet"), "{} view refreshes failed to stop in {:.3}s: {}", remaining_tasks.size(), elapsed_seconds, names);
+
+        return false;
     }
+
+    return true;
 }
 
 RefreshSet::Handle::Handle(RefreshSet * parent_set_, StorageID id_, std::optional<StorageID> inner_table_id_, RefreshTaskList::iterator iter_, RefreshTaskList::iterator inner_table_iter_, std::vector<StorageID> dependencies_)

src/Storages/MaterializedView/RefreshSet.h

@@ -68,7 +68,8 @@ class RefreshSet
     bool refreshesStopped() const;
 
     /// Called during shutdown, after setRefreshesStopped(true).
-    void joinBackgroundTasks(std::chrono::steady_clock::time_point deadline);
+    /// Returns false if some tasks are still running after deadline.
+    bool joinBackgroundTasks(std::chrono::steady_clock::time_point deadline);
 
 private:
     using TaskMap = std::unordered_map<StorageID, RefreshTaskList, StorageID::DatabaseAndTableNameHash, StorageID::DatabaseAndTableNameEqual>;

src/Storages/MaterializedView/RefreshTask.cpp

@@ -1047,7 +1047,7 @@ void RefreshTask::interruptExecution()
     if (execution.executor)
     {
         execution.executor->cancel();
-        LOG_DEBUG(log, "Cancelling refresh");
+        LOG_DEBUG(log, "Cancelling refresh in {}", set_handle.getID().getFullNameNotQuoted());
     }
 }