Merge pull request ClickHouse#73715 from vitlibar/correction-after-reworking-backup-sync-4

Corrections after reworking backup/restore synchronization #4

Author: vitlibar

src/Backups/BackupCoordinationStageSync.cpp

@@ -64,11 +64,10 @@ bool BackupCoordinationStageSync::State::operator !=(const State & other) const
 
 void BackupCoordinationStageSync::State::merge(const State & other)
 {
-    if (other.host_with_error && !host_with_error)
+    if (other.host_with_error)
     {
         const String & host = *other.host_with_error;
-        host_with_error = host;
-        hosts.at(host).exception = other.hosts.at(host).exception;
+        addErrorInfo(other.hosts.at(host).exception, host);
     }
 
     for (const auto & [host, other_host_info] : other.hosts)
@@ -81,6 +80,16 @@ void BackupCoordinationStageSync::State::merge(const State & other)
 }
 
 
+void BackupCoordinationStageSync::State::addErrorInfo(std::exception_ptr exception, const String & host)
+{
+    if (!host_with_error && exception)
+    {
+        host_with_error = host;
+        hosts.at(host).exception = exception;
+    }
+}
+
+
 BackupCoordinationStageSync::BackupCoordinationStageSync(
         bool is_restore_,
         const String & zookeeper_path_,
@@ -137,25 +146,13 @@ BackupCoordinationStageSync::BackupCoordinationStageSync(
 
 BackupCoordinationStageSync::~BackupCoordinationStageSync()
 {
-    /// Normally either finish() or setError() must be called.
-    if (!tried_to_finish)
-    {
-        if (state.host_with_error)
-        {
-            /// setError() was called and succeeded.
-            finish(/* throw_if_error = */ false);
-        }
-        else if (!tried_to_set_error)
-        {
-            /// Neither finish() nor setError() were called, it's a bug.
-            chassert(false, "~BackupCoordinationStageSync() is called without finish() or setError()");
-            LOG_ERROR(log, "~BackupCoordinationStageSync() is called without finish() or setError()");
-        }
-    }
+    /// If everything is ok, then the finish() function should be called already and the watching thread should be already stopped too.
+    /// However if an error happened then that might be different,
+    /// so here in the destructor we need to ensure that we've tried to create the finish node and also we've stopped the watching thread.
+
+    if (!finished())
+        finish(/* throw_if_error = */ false);
 
-    /// Normally the watching thread should be stopped already because the finish() function stops it.
-    /// However if an error happened then the watching thread can be still running,
-    /// so here in the destructor we have to ensure that it's stopped.
     stopWatchingThread();
 }
 
@@ -283,7 +280,11 @@ void BackupCoordinationStageSync::createStartAndAliveNodesAndCheckConcurrency(Co
         {
             auto [exception, host] = parseErrorNode(serialized_error);
             if (exception)
+            {
+                std::lock_guard lock{mutex};
+                state.addErrorInfo(exception, host);
                 std::rethrow_exception(exception);
+            }
         }
 
         if (check_concurrency)
@@ -580,12 +581,8 @@ void BackupCoordinationStageSync::readCurrentState(Coordination::ZooKeeperWithFa
             {
                 String serialized_error = zookeeper->get(error_node_path);
                 auto [exception, host] = parseErrorNode(serialized_error);
-                auto * host_info = get_host_info(host);
-                if (exception && host_info)
-                {
-                    host_info->exception = exception;
-                    new_state.host_with_error = host;
-                }
+                if (exception)
+                    new_state.addErrorInfo(exception, host);
             }
         }
         else if (zk_node.starts_with("started|"))
@@ -804,20 +801,26 @@ void BackupCoordinationStageSync::setStage(const String & stage, const String &
     {
         LOG_INFO(log, "Skipped creating the 'finish' node because the initiator uses outdated version {}", getInitiatorVersion());
         std::lock_guard lock{mutex};
-        tried_to_finish = true;
         state.hosts.at(current_host).finished = true;
     }
 }
 
 
 void BackupCoordinationStageSync::createStageNode(const String & stage, const String & stage_result, Coordination::ZooKeeperWithFaultInjection::Ptr zookeeper)
 {
+    if (isErrorSet())
+        rethrowSetError();
+
     String serialized_error;
     if (zookeeper->tryGet(error_node_path, serialized_error))
     {
         auto [exception, host] = parseErrorNode(serialized_error);
         if (exception)
+        {
+            std::lock_guard lock{mutex};
+            state.addErrorInfo(exception, host);
             std::rethrow_exception(exception);
+        }
     }
 
     auto code = zookeeper->tryCreate(getStageNodePath(stage), stage_result, zkutil::CreateMode::Persistent);
@@ -931,7 +934,7 @@ bool BackupCoordinationStageSync::finishImpl(bool throw_if_error, WithRetries::K
             return true;
         }
 
-        if (tried_to_finish)
+        if (tried_to_finish[throw_if_error])
         {
             /// We don't repeat creating the finish node, no matter if it was successful or not.
             LOG_INFO(log, "Skipped creating the finish node for {} because earlier we failed to do that", current_host_desc);
@@ -958,6 +961,11 @@ bool BackupCoordinationStageSync::finishImpl(bool throw_if_error, WithRetries::K
         }
     }
 
+    SCOPE_EXIT({
+        std::lock_guard lock{mutex};
+        tried_to_finish[throw_if_error] = true;
+    });
+
     stopWatchingThread();
 
     try
@@ -968,28 +976,18 @@ bool BackupCoordinationStageSync::finishImpl(bool throw_if_error, WithRetries::K
             with_retries.renewZooKeeper(zookeeper);
             createFinishNodeAndRemoveAliveNode(zookeeper, throw_if_error);
         });
+        return true;
     }
     catch (...)
     {
         LOG_TRACE(log, "Caught exception while creating the 'finish' node for {}: {}",
             current_host_desc,
             getCurrentExceptionMessage(/* with_stacktrace= */ false, /* check_embedded_stacktrace= */ true));
 
-        std::lock_guard lock{mutex};
-        tried_to_finish = true;
-
         if (throw_if_error)
             throw;
         return false;
     }
-
-    {
-        std::lock_guard lock{mutex};
-        tried_to_finish = true;
-        state.hosts.at(current_host).finished = true;
-    }
-
-    return true;
 }
 
 
@@ -1000,6 +998,24 @@ void BackupCoordinationStageSync::createFinishNodeAndRemoveAliveNode(Coordinatio
 
     for (size_t attempt_no = 1; attempt_no <= max_attempts_after_bad_version; ++attempt_no)
     {
+        if (throw_if_error)
+        {
+            if (isErrorSet())
+                rethrowSetError();
+
+            String serialized_error;
+            if (zookeeper->tryGet(error_node_path, serialized_error))
+            {
+                auto [exception, host] = parseErrorNode(serialized_error);
+                if (exception)
+                {
+                    std::lock_guard lock{mutex};
+                    state.addErrorInfo(exception, host);
+                    std::rethrow_exception(exception);
+                }
+            }
+        }
+
         /// The 'num_hosts' node may not exist if createStartAndAliveNodes() failed in the constructor.
         if (!num_hosts)
         {
@@ -1012,16 +1028,12 @@ void BackupCoordinationStageSync::createFinishNodeAndRemoveAliveNode(Coordinatio
             }
         }
 
-        String serialized_error;
-        if (throw_if_error && zookeeper->tryGet(error_node_path, serialized_error))
-        {
-            auto [exception, host] = parseErrorNode(serialized_error);
-            if (exception)
-                std::rethrow_exception(exception);
-        }
-
         if (zookeeper->exists(finish_node_path))
+        {
+            std::lock_guard lock{mutex};
+            state.hosts.at(current_host).finished = true;
             return;
+        }
 
         bool start_node_exists = zookeeper->exists(start_node_path);
 
@@ -1057,6 +1069,8 @@ void BackupCoordinationStageSync::createFinishNodeAndRemoveAliveNode(Coordinatio
                 hosts_left_desc = (*num_hosts == 0) ? ", no hosts left" : fmt::format(", {} hosts left", *num_hosts);
             }
             LOG_INFO(log, "Created the 'finish' node in ZooKeeper for {}{}", current_host_desc, hosts_left_desc);
+            std::lock_guard lock{mutex};
+            state.hosts.at(current_host).finished = true;
             return;
         }
 
@@ -1315,49 +1329,50 @@ bool BackupCoordinationStageSync::setError(std::exception_ptr exception, bool th
 
 bool BackupCoordinationStageSync::setError(const Exception & exception, bool throw_if_error)
 {
-    try
-    {
-        /// Most likely this exception has been already logged so here we're logging it without stacktrace.
-        String exception_message = getExceptionMessage(exception, /* with_stacktrace= */ false, /* check_embedded_stacktrace= */ true);
-        LOG_INFO(log, "Sending exception from {} to other hosts: {}", current_host_desc, exception_message);
+    /// Most likely this exception has been already logged so here we're logging it without stacktrace.
+    String exception_message = getExceptionMessage(exception, /* with_stacktrace= */ false, /* check_embedded_stacktrace= */ true);
+    LOG_INFO(log, "Sending exception from {} to other hosts: {}", current_host_desc, exception_message);
 
+    {
+        std::lock_guard lock{mutex};
+        if (state.host_with_error)
         {
-            std::lock_guard lock{mutex};
-            if (state.host_with_error)
-            {
-                LOG_INFO(log, "The error node already exists");
-                return true;
-            }
+            /// We create the error node always before assigning `state.host_with_error`,
+            /// thus if `state.host_with_error` is set then we can be sure that the error node exists.
+            LOG_INFO(log, "The error node already exists");
+            return true;
+        }
 
-            if (tried_to_set_error)
-            {
-                LOG_INFO(log, "Skipped creating the error node because earlier we failed to do that");
-                return false;
-            }
+        if (tried_to_set_error)
+        {
+            LOG_INFO(log, "Skipped creating the error node because earlier we failed to do that");
+            return false;
         }
+    }
+
+    SCOPE_EXIT({
+        std::lock_guard lock{mutex};
+        tried_to_set_error = true;
+    });
+
+    try
+    {
+
 
         auto holder = with_retries.createRetriesControlHolder("BackupCoordinationStageSync::setError", WithRetries::kErrorHandling);
         holder.retries_ctl.retryLoop([&, &zookeeper = holder.faulty_zookeeper]()
         {
             with_retries.renewZooKeeper(zookeeper);
             createErrorNode(exception, zookeeper);
         });
-
-        {
-            std::lock_guard lock{mutex};
-            tried_to_set_error = true;
-            return true;
-        }
+        return true;
     }
     catch (...)
     {
-        LOG_TRACE(log, "Caught exception while removing nodes from ZooKeeper for this {}: {}",
+        LOG_TRACE(log, "Caught exception while creating the error node for this {}: {}",
                   is_restore ? "restore" : "backup",
                   getCurrentExceptionMessage(/* with_stacktrace= */ false, /* check_embedded_stacktrace= */ true));
 
-        std::lock_guard lock{mutex};
-        tried_to_set_error = true;
-
         if (throw_if_error)
             throw;
         return false;
@@ -1367,6 +1382,14 @@ bool BackupCoordinationStageSync::setError(const Exception & exception, bool thr
 
 void BackupCoordinationStageSync::createErrorNode(const Exception & exception, Coordination::ZooKeeperWithFaultInjection::Ptr zookeeper)
 {
+    if (isErrorSet())
+    {
+        /// We create the error node always before assigning `state.host_with_error`,
+        /// thus if `state.host_with_error` is set then we can be sure that the error node exists.
+        LOG_INFO(log, "The error node already exists");
+        return;
+    }
+
     String serialized_error;
     {
         WriteBufferFromOwnString buf;
@@ -1375,16 +1398,15 @@ void BackupCoordinationStageSync::createErrorNode(const Exception & exception, C
         serialized_error = buf.str();
     }
 
+    zookeeper->createIfNotExists(operation_zookeeper_path, "");
+    zookeeper->createIfNotExists(zookeeper_path, "");
+
     auto code = zookeeper->tryCreate(error_node_path, serialized_error, zkutil::CreateMode::Persistent);
 
     if (code == Coordination::Error::ZOK)
     {
         std::lock_guard lock{mutex};
-        if (!state.host_with_error)
-        {
-            state.host_with_error = current_host;
-            state.hosts.at(current_host).exception = parseErrorNode(serialized_error).first;
-        }
+        state.addErrorInfo(parseErrorNode(serialized_error).first, current_host);
         LOG_TRACE(log, "Sent exception from {} to other hosts", current_host_desc);
         return;
     }
@@ -1396,11 +1418,7 @@ void BackupCoordinationStageSync::createErrorNode(const Exception & exception, C
         if (another_exception)
         {
             std::lock_guard lock{mutex};
-            if (!state.host_with_error)
-            {
-                state.host_with_error = host;
-                state.hosts.at(host).exception = another_exception;
-            }
+            state.addErrorInfo(another_exception, host);
             LOG_INFO(log, "Another error is already assigned for this {}", operation_name);
             return;
         }
@@ -1428,4 +1446,11 @@ bool BackupCoordinationStageSync::isErrorSet() const
     return state.host_with_error.has_value();
 }
 
+void BackupCoordinationStageSync::rethrowSetError() const
+{
+    std::lock_guard lock{mutex};
+    chassert(state.host_with_error);
+    std::rethrow_exception(state.hosts.at(*state.host_with_error).exception);
+}
+
 }

src/Backups/BackupCoordinationStageSync.h

@@ -98,6 +98,7 @@ class BackupCoordinationStageSync
 
     /// Lets other hosts know that the current host has encountered an error.
     bool setError(const Exception & exception, bool throw_if_error);
+    [[noreturn]] void rethrowSetError() const;
     void createErrorNode(const Exception & exception, Coordination::ZooKeeperWithFaultInjection::Ptr zookeeper);
 
     /// Deserializes an error stored in the error node.
@@ -207,7 +208,9 @@ class BackupCoordinationStageSync
 
         bool operator ==(const State & other) const;
         bool operator !=(const State & other) const;
+
         void merge(const State & other);
+        void addErrorInfo(std::exception_ptr exception, const String & host);
     };
 
     State state TSA_GUARDED_BY(mutex);
@@ -217,7 +220,7 @@ class BackupCoordinationStageSync
     bool should_stop_watching_thread TSA_GUARDED_BY(mutex) = false;
 
     bool query_is_sent_to_other_hosts TSA_GUARDED_BY(mutex) = false;
-    bool tried_to_finish TSA_GUARDED_BY(mutex) = false;
+    bool tried_to_finish[2] TSA_GUARDED_BY(mutex) = {false, false};
     bool tried_to_set_error TSA_GUARDED_BY(mutex) = false;
 
     mutable std::mutex mutex;