fix: eliminate deadlocks by separating node_states and callback mutexes

Use std::atomic<bool> for is_connected_ and primary_host_online_ in
EdgeNode so Paho callbacks never need the application mutex. Add a
dedicated node_states_mutex_ in HostApplication to protect node_states_
independently, allowing validate_message to call log() directly without
the PendingLogs indirection. Phase connect()/disconnect() to release
mutex_ before blocking waits. Fix missing primary_host_online_ transfer
in EdgeNode move assignment. Safer destructors that always attempt
disconnect.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: Jan van Lindt

include/sparkplug/edge_node.hpp

@@ -7,6 +7,7 @@
 #include "sparkplug_b.pb.h"
 #include "topic.hpp"
 
+#include <atomic>
 #include <functional>
 #include <mutex>
 #include <optional>
@@ -465,8 +466,8 @@ class EdgeNode {
   // Track state of attached devices (device_id -> state, with heterogeneous lookup)
   std::unordered_map<std::string, DeviceState, StringHash, StringEqual> device_states_;
 
-  bool is_connected_{false};
-  bool primary_host_online_{
+  std::atomic<bool> is_connected_{false};
+  std::atomic<bool> primary_host_online_{
       false}; // True if primary host is online (or no primary host configured)
 
   // Mutex for thread-safe access to all mutable state

include/sparkplug/host_application.hpp

@@ -7,6 +7,7 @@
 #include "sparkplug_b.pb.h"
 #include "topic.hpp"
 
+#include <atomic>
 #include <functional>
 #include <memory>
 #include <mutex>
@@ -492,7 +493,7 @@ class HostApplication {
 private:
   Config config_;
   MQTTAsyncHandle client_;
-  bool is_connected_{false};
+  std::atomic<bool> is_connected_{false};
 
   // MQTT connection options that must outlive async operations
   MQTTAsync_SSLOptions ssl_opts_{};
@@ -539,8 +540,9 @@ class HostApplication {
   };
 
   std::unordered_map<NodeKey, NodeState, NodeKeyHash, NodeKeyEqual> node_states_;
+  mutable std::mutex node_states_mutex_; // Protects node_states_ only
 
-  // Mutex for thread-safe access to all mutable state
+  // Mutex for thread-safe access to config and other mutable state
   mutable std::mutex mutex_;
 
   [[nodiscard]] stdx::expected<void, std::string>

src/c_bindings.cpp

@@ -3,6 +3,7 @@
 #include "sparkplug/payload_builder.hpp"
 #include "sparkplug/sparkplug_c.h"
 
+#include <chrono>
 #include <cstring>
 #include <format>
 
@@ -34,6 +35,12 @@ copy_metrics_to_builder(sparkplug::PayloadBuilder& builder,
     const char* name = metric.has_name() ? metric.name().c_str() : "";
     std::optional<uint64_t> alias =
         metric.has_alias() ? std::optional<uint64_t>(metric.alias()) : std::nullopt;
+    uint64_t ts = metric.has_timestamp()
+                      ? metric.timestamp()
+                      : static_cast<uint64_t>(
+                            std::chrono::duration_cast<std::chrono::milliseconds>(
+                                std::chrono::system_clock::now().time_since_epoch())
+                                .count());
 
     auto datatype = static_cast<sparkplug::DataType>(metric.datatype());
 
@@ -42,87 +49,89 @@ copy_metrics_to_builder(sparkplug::PayloadBuilder& builder,
     case sparkplug::DataType::Int16:
     case sparkplug::DataType::Int32:
       if (alias.has_value() && name[0] == '\0') {
-        builder.add_metric_by_alias(*alias, static_cast<int32_t>(metric.int_value()));
+        builder.add_metric_by_alias(*alias, static_cast<int32_t>(metric.int_value()), ts);
       } else if (alias.has_value()) {
         builder.add_metric_with_alias(name, *alias,
-                                      static_cast<int32_t>(metric.int_value()));
+                                      static_cast<int32_t>(metric.int_value()), ts);
       } else {
-        builder.add_metric(name, static_cast<int32_t>(metric.int_value()));
+        builder.add_metric(name, static_cast<int32_t>(metric.int_value()), ts);
       }
       break;
 
     case sparkplug::DataType::Int64:
       if (alias.has_value() && name[0] == '\0') {
-        builder.add_metric_by_alias(*alias, metric.long_value());
+        builder.add_metric_by_alias(*alias, metric.long_value(), ts);
       } else if (alias.has_value()) {
-        builder.add_metric_with_alias(name, *alias, metric.long_value());
+        builder.add_metric_with_alias(name, *alias, metric.long_value(), ts);
       } else {
-        builder.add_metric(name, metric.long_value());
+        builder.add_metric(name, metric.long_value(), ts);
       }
       break;
 
     case sparkplug::DataType::UInt8:
     case sparkplug::DataType::UInt16:
     case sparkplug::DataType::UInt32:
       if (alias.has_value() && name[0] == '\0') {
-        builder.add_metric_by_alias(*alias, static_cast<uint32_t>(metric.int_value()));
+        builder.add_metric_by_alias(*alias, static_cast<uint32_t>(metric.int_value()),
+                                    ts);
       } else if (alias.has_value()) {
         builder.add_metric_with_alias(name, *alias,
-                                      static_cast<uint32_t>(metric.int_value()));
+                                      static_cast<uint32_t>(metric.int_value()), ts);
       } else {
-        builder.add_metric(name, static_cast<uint32_t>(metric.int_value()));
+        builder.add_metric(name, static_cast<uint32_t>(metric.int_value()), ts);
       }
       break;
 
     case sparkplug::DataType::UInt64:
       if (alias.has_value() && name[0] == '\0') {
-        builder.add_metric_by_alias(*alias, static_cast<uint64_t>(metric.long_value()));
+        builder.add_metric_by_alias(*alias, static_cast<uint64_t>(metric.long_value()),
+                                    ts);
       } else if (alias.has_value()) {
         builder.add_metric_with_alias(name, *alias,
-                                      static_cast<uint64_t>(metric.long_value()));
+                                      static_cast<uint64_t>(metric.long_value()), ts);
       } else {
-        builder.add_metric(name, static_cast<uint64_t>(metric.long_value()));
+        builder.add_metric(name, static_cast<uint64_t>(metric.long_value()), ts);
       }
       break;
 
     case sparkplug::DataType::Float:
       if (alias.has_value() && name[0] == '\0') {
-        builder.add_metric_by_alias(*alias, metric.float_value());
+        builder.add_metric_by_alias(*alias, metric.float_value(), ts);
       } else if (alias.has_value()) {
-        builder.add_metric_with_alias(name, *alias, metric.float_value());
+        builder.add_metric_with_alias(name, *alias, metric.float_value(), ts);
       } else {
-        builder.add_metric(name, metric.float_value());
+        builder.add_metric(name, metric.float_value(), ts);
       }
       break;
 
     case sparkplug::DataType::Double:
       if (alias.has_value() && name[0] == '\0') {
-        builder.add_metric_by_alias(*alias, metric.double_value());
+        builder.add_metric_by_alias(*alias, metric.double_value(), ts);
       } else if (alias.has_value()) {
-        builder.add_metric_with_alias(name, *alias, metric.double_value());
+        builder.add_metric_with_alias(name, *alias, metric.double_value(), ts);
       } else {
-        builder.add_metric(name, metric.double_value());
+        builder.add_metric(name, metric.double_value(), ts);
       }
       break;
 
     case sparkplug::DataType::Boolean:
       if (alias.has_value() && name[0] == '\0') {
-        builder.add_metric_by_alias(*alias, metric.boolean_value());
+        builder.add_metric_by_alias(*alias, metric.boolean_value(), ts);
       } else if (alias.has_value()) {
-        builder.add_metric_with_alias(name, *alias, metric.boolean_value());
+        builder.add_metric_with_alias(name, *alias, metric.boolean_value(), ts);
       } else {
-        builder.add_metric(name, metric.boolean_value());
+        builder.add_metric(name, metric.boolean_value(), ts);
       }
       break;
 
     case sparkplug::DataType::String:
     case sparkplug::DataType::Text:
       if (alias.has_value() && name[0] == '\0') {
-        builder.add_metric_by_alias(*alias, metric.string_value());
+        builder.add_metric_by_alias(*alias, metric.string_value(), ts);
       } else if (alias.has_value()) {
-        builder.add_metric_with_alias(name, *alias, metric.string_value());
+        builder.add_metric_with_alias(name, *alias, metric.string_value(), ts);
       } else {
-        builder.add_metric(name, metric.string_value());
+        builder.add_metric(name, metric.string_value(), ts);
       }
       break;