Backport #73323 to 24.12: Fix refreshable MV crash on shutdown

robot-clickhouse · robot-clickhouse · commit a7790a8b25c0 · 2025-01-20T18:08:59.000Z
diff --git a/programs/server/Server.cpp b/programs/server/Server.cpp
@@ -2523,6 +2523,8 @@ try
                 }
             }
 
+            global_context->getRefreshSet().setRefreshesStopped(true);
+
             if (current_connections)
                 LOG_WARNING(log, "Closed all listening sockets. Waiting for {} outstanding connections.", current_connections);
             else
@@ -2538,15 +2540,20 @@ try
             if (!server_settings[ServerSetting::shutdown_wait_unfinished_queries])
                 global_context->getProcessList().killAllQueries();
 
+            size_t wait_limit_seconds = server_settings[ServerSetting::shutdown_wait_unfinished];
+            auto wait_start = std::chrono::steady_clock::now();
+
             if (current_connections)
-                current_connections = waitServersToFinish(servers, servers_lock, server_settings[ServerSetting::shutdown_wait_unfinished]);
+                current_connections = waitServersToFinish(servers, servers_lock, wait_limit_seconds);
 
             if (current_connections)
                 LOG_WARNING(log, "Closed connections. But {} remain."
                     " Tip: To increase wait time add to config: <shutdown_wait_unfinished>60</shutdown_wait_unfinished>", current_connections);
             else
                 LOG_INFO(log, "Closed connections.");
 
+            global_context->getRefreshSet().joinBackgroundTasks(wait_start + std::chrono::milliseconds(wait_limit_seconds * 1000));
+
             dns_cache_updater.reset();
 
             if (current_connections)
diff --git a/src/Databases/DatabaseReplicatedWorker.cpp b/src/Databases/DatabaseReplicatedWorker.cpp
@@ -334,21 +334,28 @@ String DatabaseReplicatedDDLWorker::tryEnqueueAndExecuteEntry(DDLLogEntry & entr
     assert(!zookeeper->exists(task->getFinishedNodePath()));
     task->is_initial_query = true;
 
-    LOG_DEBUG(log, "Waiting for worker thread to process all entries before {}", entry_name);
     UInt64 timeout = query_context->getSettingsRef()[Setting::database_replicated_initial_query_timeout_sec];
     StopToken cancellation = query_context->getDDLQueryCancellation();
     StopCallback cancellation_callback(cancellation, [&] { wait_current_task_change.notify_all(); });
+    LOG_DEBUG(log, "Waiting for worker thread to process all entries before {} (timeout: {}s{})", entry_name, timeout, cancellation.stop_possible() ? ", cancellable" : "");
+
     {
         std::unique_lock lock{mutex};
         bool processed = wait_current_task_change.wait_for(lock, std::chrono::seconds(timeout), [&]()
         {
             assert(zookeeper->expired() || current_task <= entry_name);
 
             if (zookeeper->expired() || stop_flag)
+            {
+                LOG_TRACE(log, "Not enqueueing query: {}", stop_flag ? "replication stopped" : "ZooKeeper session expired");
                 throw Exception(ErrorCodes::DATABASE_REPLICATION_FAILED, "ZooKeeper session expired or replication stopped, try again");
+            }
 
             if (cancellation.stop_requested())
+            {
+                LOG_TRACE(log, "DDL query was cancelled");
                 throw Exception(ErrorCodes::QUERY_WAS_CANCELLED, "DDL query was cancelled");
+            }
 
             return current_task == entry_name;
         });
diff --git a/src/Storages/MaterializedView/RefreshSet.cpp b/src/Storages/MaterializedView/RefreshSet.cpp
@@ -185,7 +185,10 @@ void RefreshSet::setRefreshesStopped(bool stopped)
     TaskMap tasks_copy;
     {
         std::lock_guard lock(mutex);
-        refreshes_stopped.store(stopped);
+        if (refreshes_stopped.exchange(stopped) == stopped)
+            return;
+        if (stopped)
+            refreshes_stopped_at = std::chrono::steady_clock::now();
         tasks_copy = tasks;
     }
     for (const auto & kv : tasks_copy)
@@ -198,6 +201,40 @@ bool RefreshSet::refreshesStopped() const
     return refreshes_stopped.load();
 }
 
+void RefreshSet::joinBackgroundTasks(std::chrono::steady_clock::time_point deadline)
+{
+    std::vector<RefreshTaskPtr> remaining_tasks;
+    std::chrono::steady_clock::time_point stopped_at;
+    {
+        std::unique_lock lock(mutex);
+        stopped_at = refreshes_stopped_at;
+        for (const auto & [_, list] : tasks)
+            remaining_tasks.insert(remaining_tasks.end(), list.begin(), list.end());
+    }
+    std::erase_if(remaining_tasks, [&](const auto & t)
+        {
+            return t->tryJoinBackgroundTask(deadline);
+        });
+
+    if (!remaining_tasks.empty())
+    {
+        auto elapsed_seconds = std::chrono::duration_cast<std::chrono::duration<double>>(std::chrono::steady_clock::now() - stopped_at).count();
+        String names;
+        for (size_t i = 0; i < remaining_tasks.size(); ++i)
+        {
+            if (i > 0)
+                names += ", ";
+            if (i >= 20)
+            {
+                names += "...";
+                break;
+            }
+            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);
+    }
+}
+
 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_)
     : parent_set(parent_set_), id(std::move(id_)), inner_table_id(std::move(inner_table_id_)), dependencies(std::move(dependencies_))
     , iter(iter_), inner_table_iter(inner_table_iter_), metric_increment(CurrentMetrics::Increment(CurrentMetrics::RefreshableViews)) {}
diff --git a/src/Storages/MaterializedView/RefreshSet.h b/src/Storages/MaterializedView/RefreshSet.h
@@ -68,6 +68,9 @@ class RefreshSet
     void setRefreshesStopped(bool stopped);
     bool refreshesStopped() const;
 
+    /// Called during shutdown, after setRefreshesStopped(true).
+    void joinBackgroundTasks(std::chrono::steady_clock::time_point deadline);
+
 private:
     using TaskMap = std::unordered_map<StorageID, RefreshTaskList, StorageID::DatabaseAndTableNameHash, StorageID::DatabaseAndTableNameEqual>;
     using DependentsMap = std::unordered_map<StorageID, std::unordered_set<RefreshTaskPtr>, StorageID::DatabaseAndTableNameHash, StorageID::DatabaseAndTableNameEqual>;
@@ -82,6 +85,7 @@ class RefreshSet
     InnerTableMap inner_tables;
 
     std::atomic<bool> refreshes_stopped {false};
+    std::chrono::steady_clock::time_point refreshes_stopped_at;
 
     RefreshTaskList::iterator addTaskLocked(StorageID id, RefreshTaskPtr task);
     void removeTaskLocked(StorageID id, RefreshTaskList::iterator iter);
diff --git a/src/Storages/MaterializedView/RefreshTask.cpp b/src/Storages/MaterializedView/RefreshTask.cpp
@@ -307,6 +307,22 @@ void RefreshTask::wait()
     }
 }
 
+bool RefreshTask::tryJoinBackgroundTask(std::chrono::steady_clock::time_point deadline)
+{
+    std::unique_lock lock(mutex);
+
+    execution.cancel_ddl_queries.request_stop();
+
+    auto duration = deadline - std::chrono::steady_clock::now();
+    /// (Manually clamping to 0 because the standard library used to have (and possibly still has?)
+    ///  a bug that wait_until would wait forever if the timestamp is in the past.)
+    duration = std::max(duration, std::chrono::steady_clock::duration(0));
+    return refresh_cv.wait_for(lock, duration, [&]
+        {
+            return state != RefreshState::Running && state != RefreshState::Scheduling;
+        });
+}
+
 std::chrono::sys_seconds RefreshTask::getNextRefreshTimeslot() const
 {
     std::lock_guard guard(mutex);
diff --git a/src/Storages/MaterializedView/RefreshTask.h b/src/Storages/MaterializedView/RefreshTask.h
@@ -75,11 +75,16 @@ class RefreshTask : public std::enable_shared_from_this<RefreshTask>
     /// Cancel task execution
     void cancel();
 
-    /// Waits for the currently running refresh attempt to complete.
+    /// Waits for the currently running refresh attempt to complete, either on this replica
+    /// or on another one (if `coordinated`).
     /// If the refresh fails, throws an exception.
     /// If no refresh is running, completes immediately, throwing an exception if previous refresh failed.
     void wait();
 
+    /// Wait for background work (refreshing or scheduling) on this replica to complete.
+    /// Returns false if `deadline` was reached before the work completed. Used by server shutdown.
+    bool tryJoinBackgroundTask(std::chrono::steady_clock::time_point deadline);
+
     /// A measure of how far this view has progressed. Used by dependent views.
     std::chrono::sys_seconds getNextRefreshTimeslot() const;
 
diff --git a/tests/integration/helpers/cluster.py b/tests/integration/helpers/cluster.py
@@ -2047,12 +2047,12 @@ def _replace(self, path, what, to):
     def restart_instance_with_ip_change(self, node, new_ip):
         if "::" in new_ip:
             if node.ipv6_address is None:
-                raise Exception("You should specity ipv6_address in add_node method")
+                raise Exception("You should specify ipv6_address in add_node method")
             self._replace(node.docker_compose_path, node.ipv6_address, new_ip)
             node.ipv6_address = new_ip
         else:
             if node.ipv4_address is None:
-                raise Exception("You should specity ipv4_address in add_node method")
+                raise Exception("You should specify ipv4_address in add_node method")
             self._replace(node.docker_compose_path, node.ipv4_address, new_ip)
             node.ipv4_address = new_ip
         run_and_check(self.base_cmd + ["stop", node.name])
@@ -2119,9 +2119,11 @@ def exec_in_container(
         detach: bool = False,
         nothrow: bool = False,
         use_cli: bool = True,
+        get_exec_id: bool = False,
         **kwargs: Any,
     ) -> str:
         if use_cli:
+            assert not get_exec_id
             logging.debug(
                 f"run container_id:{container_id} detach:{detach} nothrow:{nothrow} cmd: {cmd}"
             )
@@ -2159,8 +2161,9 @@ def exec_in_container(
                 else:
                     raise Exception(message)
             if not detach:
+                assert not get_exec_id
                 return output.decode()
-            return output
+            return exec_id if get_exec_id else output
 
     def copy_file_to_container(self, container_id, local_path, dest_path):
         with open(local_path, "rb") as fdata:
@@ -2732,7 +2735,7 @@ def wait_cassandra_to_start(self, timeout=180):
                 logging.info(
                     f"Check Cassandra Online {self.cassandra_id} {self.cassandra_ip} {self.cassandra_port}"
                 )
-                check = self.exec_in_container(
+                self.exec_in_container(
                     self.cassandra_id,
                     [
                         "bash",
@@ -4032,16 +4035,20 @@ def start_clickhouse(
             )
         start_time = time.time()
         time_to_sleep = 0.5
+        exec_id = None
 
         while start_time + start_wait_sec >= time.time():
             # sometimes after SIGKILL (hard reset) server may refuse to start for some time
             # for different reasons.
             pid = self.get_process_pid("clickhouse")
             if pid is None:
                 logging.debug("No clickhouse process running. Start new one.")
-                self.exec_in_container(
-                    ["bash", "-c", self.clickhouse_start_command_in_daemon],
+                exec_id = self.exec_in_container(
+                    ["bash", "-c", self.clickhouse_start_command],
                     user=str(os.getuid()),
+                    detach=True,
+                    use_cli=False,
+                    get_exec_id=True,
                 )
                 if expected_to_fail:
                     self.wait_start_failed(start_wait_sec + start_time - time.time())
@@ -4054,7 +4061,7 @@ def start_clickhouse(
                     raise Exception("ClickHouse was expected not to be running.")
                 try:
                     self.wait_start(start_wait_sec + start_time - time.time())
-                    return
+                    return exec_id
                 except Exception as e:
                     logging.warning(
                         f"Current start attempt failed. Will kill {pid} just in case."
diff --git a/tests/integration/test_refreshable_mv/test.py b/tests/integration/test_refreshable_mv/test.py
@@ -1,7 +1,4 @@
-import os
-import re
-import shutil
-import threading
+import logging
 import time
 from random import randint
 
@@ -221,3 +218,55 @@ def test_refreshable_mv_in_system_db(started_cluster):
     assert node1.query("select count(), sum(x) from system.a") == "2\t3\n"
 
     node1.query("drop table system.a")
+
+
+def test_refresh_vs_shutdown_smoke(started_cluster):
+    for node in nodes:
+        node.query(
+            "create database re engine = Replicated('/test/re', 'shard1', '{replica}');"
+        )
+
+    node1.stop_clickhouse()
+
+    num_tables = 2
+
+    for i in range(10):
+        exec_id = node1.start_clickhouse()
+        assert exec_id is not None
+
+        if i == 0:
+            node1.query("select '===test_refresh_vs_shutdown_smoke start==='")
+            for j in range(num_tables):
+                node1.query(
+                    f"create materialized view re.a{j} refresh every 1 second (x Int64) engine ReplicatedMergeTree order by x as select number*10 as x from numbers(2)"
+                )
+
+        if randint(0, 1):
+            for j in range(num_tables):
+                if randint(0, 1):
+                    node1.query(f"system refresh view re.a{j}")
+        r = randint(0, 2)
+        if r == 1:
+            time.sleep(randint(0, 10) / 1000)
+        elif r == 2:
+            time.sleep(randint(0, 100) / 1000)
+
+        node1.stop_clickhouse(stop_wait_sec=300)
+        while True:
+            exit_code = cluster.docker_client.api.exec_inspect(exec_id)["ExitCode"]
+            if exit_code is not None:
+                assert exit_code == 0
+                break
+            time.sleep(1)
+
+    assert not node1.contains_in_log("view refreshes failed to stop", from_host=True)
+    assert not node1.contains_in_log("Closed connections. But", from_host=True)
+    assert not node1.contains_in_log("Will shutdown forcefully.", from_host=True)
+    assert not node1.contains_in_log("##########", from_host=True)
+    assert node1.contains_in_log(
+        "===test_refresh_vs_shutdown_smoke start===", from_host=True
+    )
+
+    node1.start_clickhouse()
+    node1.query("drop database re sync")
+    node2.query("drop database re sync")
