MQTTManager: don't let new messages overtake ones buffered while disconnected

docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp

@@ -119,13 +119,60 @@ void MQTT_Manager::reload_config() {
 }
 
 bool MQTT_Manager::is_connected() {
+  return MQTT_Manager::_caught_up && MQTT_Manager::_broker_connected();
+}
+
+bool MQTT_Manager::_broker_connected() {
+#if defined(TEST_MODE) && TEST_MODE == 1
+  if (MQTT_Manager::_test_broker_connected) {
+    return true;
+  }
+#endif
   if (MQTT_Manager::_mqtt_client != nullptr) {
     return MQTT_Manager::_mqtt_client->is_connected();
   } else {
     return false;
   }
 }
 
+void MQTT_Manager::_send(mqtt::const_message_ptr msg) {
+#if defined(TEST_MODE) && TEST_MODE == 1
+  if (MQTT_Manager::_test_broker_connected) {
+    std::lock_guard<std::mutex> lock_guard(MQTT_Manager::_test_published_messages_mutex);
+    MQTT_Manager::_test_sent_messages.push_back({msg->get_topic(), msg->get_payload_str(), msg->is_retained()});
+    return;
+  }
+#endif
+  MQTT_Manager::_mqtt_client->publish(msg);
+}
+
+void MQTT_Manager::_send_buffered_messages() {
+  std::lock_guard<std::mutex> mutex_guard(MQTT_Manager::_mqtt_client_mutex);
+  try {
+    auto it = MQTT_Manager::_mqtt_messages_buffer.cbegin();
+    while (it != MQTT_Manager::_mqtt_messages_buffer.cend()) {
+      MQTT_Manager::_send(*it);
+      MQTT_Manager::_mqtt_messages_buffer.erase(it++);
+    }
+
+    // Then the latest value of every retained topic, as the server may have restarted and lost them.
+    // Sent last so that each topic ends on its newest value, whatever was buffered for it.
+    std::unordered_map<std::string, std::string> retained_messages;
+    {
+      std::lock_guard<std::mutex> retain_buffer_guard(MQTT_Manager::_mqtt_retain_buffer_mutex);
+      retained_messages = MQTT_Manager::_mqtt_retain_buffer;
+    }
+    for (const auto &[topic, payload] : retained_messages) {
+      MQTT_Manager::_send(mqtt::make_message(topic, payload, 0, true));
+    }
+  } catch (std::exception &e) {
+    SPDLOG_ERROR("Caught exception when trying to publish message: {}", boost::diagnostic_information(e, true));
+  }
+  // Even if a send failed: buffering everything from now on would be worse, and losing the
+  // connection again starts over.
+  MQTT_Manager::_caught_up = true;
+}
+
 void MQTT_Manager::_reconnect_mqtt_client() {
   WebsocketServer::register_warning(WebsocketServer::ActiveWarningLevel::ERROR, "MQTT not connected.");
 
@@ -163,6 +210,8 @@ void MQTT_Manager::_reconnect_mqtt_client() {
   }
 
   MQTT_Manager::_stop_consuming = true;
+  // Buffer everything published from here until the buffered messages have been sent after connecting.
+  MQTT_Manager::_caught_up = false;
   while (true) {
     try {
       std::lock_guard<std::mutex> lock_guard(MQTT_Manager::_mqtt_client_mutex);
@@ -217,16 +266,9 @@ void MQTT_Manager::_reconnect_mqtt_client() {
   MQTT_Manager::publish(fmt::format("nspanel/mqttmanager_{}/status/status", MqttManagerConfig::get_setting_with_default<std::string>(MQTT_MANAGER_SETTING::MANAGER_ADDRESS)),
                         "online", true);
 
-  // Loop over retained MQTT messages and send them again to the MQTT broker as it may have restarted and lost retained messages.
-  // Publish from a copy, as publish() writes to the buffer.
-  std::unordered_map<std::string, std::string> retained_messages;
-  {
-    std::lock_guard<std::mutex> retain_buffer_guard(MQTT_Manager::_mqtt_retain_buffer_mutex);
-    retained_messages = MQTT_Manager::_mqtt_retain_buffer;
-  }
-  for (auto it = retained_messages.cbegin(); it != retained_messages.cend(); it++) {
-    MQTT_Manager::publish(it->first, it->second, true);
-  }
+  // Before subscribing, which takes a while: until this is done everything published is buffered
+  // behind what was buffered while disconnected.
+  MQTT_Manager::_send_buffered_messages();
 
   try {
     SPDLOG_DEBUG("Subscribing to registered MQTT topics.");
@@ -253,18 +295,6 @@ void MQTT_Manager::_reconnect_mqtt_client() {
   } catch (std::exception &e) {
     SPDLOG_ERROR("Caught exception trying to subscribe to topics: {}", boost::diagnostic_information(e, true));
   }
-
-  // Send buffered messages if any
-  try {
-    std::lock_guard<std::mutex> mutex_guard(MQTT_Manager::_mqtt_client_mutex);
-    auto it = MQTT_Manager::_mqtt_messages_buffer.cbegin();
-    while (it != MQTT_Manager::_mqtt_messages_buffer.cend()) {
-      MQTT_Manager::_mqtt_client->publish((*it));
-      MQTT_Manager::_mqtt_messages_buffer.erase(it++);
-    }
-  } catch (std::exception &e) {
-    SPDLOG_ERROR("Caught exception when trying to publish message: {}", boost::diagnostic_information(e, true));
-  }
 }
 
 void MQTT_Manager::_process_mqtt_messages() {
@@ -318,7 +348,7 @@ void MQTT_Manager::publish(const std::string &topic, const std::string &payload,
       SPDLOG_TRACE("Publising '{}' -> '{}'", topic, payload);
       while (MQTT_Manager::is_connected()) {
         try {
-          MQTT_Manager::_mqtt_client->publish(msg);
+          MQTT_Manager::_send(msg);
           break;
         } catch (std::exception &ex) {
           SPDLOG_WARN("Caught exception while trying to publish '{}' -> '{}'. Retrying...", topic, payload);
@@ -370,8 +400,9 @@ void MQTT_Manager::clear_retain(const std::string &topic) {
   WebsocketServer::set_stomp_topic_retained(fmt::format("mqtt/{}", topic), false);
   if (topic.size() > 0) {
     mqtt::message_ptr msg = mqtt::make_message(topic.c_str(), "", 0, 0, true);
-    if (MQTT_Manager::_mqtt_client != nullptr && MQTT_Manager::_mqtt_client->is_connected()) {
-      MQTT_Manager::_mqtt_client->publish(msg);
+    // Through the same gate as publish(), so that the clear cannot overtake a buffered value for the topic.
+    if (MQTT_Manager::_mqtt_client != nullptr && MQTT_Manager::is_connected()) {
+      MQTT_Manager::_send(msg);
     } else {
       MQTT_Manager::_mqtt_messages_buffer.push_back(msg);
     }

docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp

@@ -37,6 +37,11 @@ class MQTT_Manager {
    */
   static void reload_config();
 
+  /*
+   * Whether messages are sent to the MQTT server as they are published. False from losing the
+   * connection until, after reconnecting, the messages buffered meanwhile have been sent: until then
+   * new messages are buffered behind them, so that nothing overtakes an older message.
+   */
   static bool is_connected();
 
   template <typename CALLBACK_BIND>
@@ -129,12 +134,35 @@ class MQTT_Manager {
     std::lock_guard<std::mutex> mutex_guard(MQTT_Manager::_mqtt_client_mutex);
     return MQTT_Manager::_mqtt_callbacks.count(topic) > 0 ? MQTT_Manager::_mqtt_callbacks.at(topic).num_slots() : 0;
   }
+
+  // While set, behave as if connected to the MQTT server, but record what would have been sent to it
+  // rather than sending it. Clearing it behaves as losing the connection.
+  static void test_set_broker_connected(bool connected) {
+    MQTT_Manager::_test_broker_connected = connected;
+    if (!connected) {
+      MQTT_Manager::_caught_up = false;
+    }
+  }
+
+  // Send what was buffered while disconnected, as after reconnecting.
+  static void test_send_buffered_messages() {
+    MQTT_Manager::_send_buffered_messages();
+  }
+
+  // Everything sent to the MQTT server since the last call, in order. Only recorded while
+  // test_set_broker_connected(true).
+  static std::vector<TestPublishedMessage> test_take_sent_messages() {
+    std::lock_guard<std::mutex> lock_guard(MQTT_Manager::_test_published_messages_mutex);
+    return std::exchange(MQTT_Manager::_test_sent_messages, {});
+  }
 #endif
 
 private:
 #if defined(TEST_MODE) && TEST_MODE == 1
   static inline std::mutex _test_published_messages_mutex;
   static inline std::vector<TestPublishedMessage> _test_published_messages;
+  static inline std::vector<TestPublishedMessage> _test_sent_messages;
+  static inline std::atomic<bool> _test_broker_connected = false;
 #endif
   static inline boost::lockfree::spsc_queue<MQTTMessage, boost::lockfree::capacity<256>> _mqtt_message_queue;
   static inline std::thread _process_messages_thread;
@@ -147,6 +175,22 @@ class MQTT_Manager {
 
   static void _reconnect_mqtt_client();
 
+  // Whether the MQTT client is connected to the server, regardless of _caught_up.
+  static bool _broker_connected();
+
+  // Set once the messages buffered while disconnected have been sent, see is_connected().
+  static inline std::atomic<bool> _caught_up = false;
+
+  /**
+   * Send the messages buffered while disconnected, in the order they were published, then the latest
+   * value of every retained topic in case the server restarted and lost them, and mark the client as
+   * caught up. Holds the client lock throughout, so nothing published meanwhile can overtake them.
+   */
+  static void _send_buffered_messages();
+
+  // Hand a message to the MQTT client. Called with _mqtt_client_mutex held.
+  static void _send(mqtt::const_message_ptr msg);
+
   static inline boost::ptr_map<std::string, boost::signals2::signal<void(std::string, std::string)>> _mqtt_callbacks;
   static inline std::unordered_map<std::string, int> _subscribed_topics;
 

docker/MQTTManager/tests/main.cpp

@@ -2,6 +2,7 @@
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
+#include "test_helpers.hpp"
 #include <cxxabi.h>
 #include <database_manager/database_manager.hpp>
 #include <execinfo.h>
@@ -57,6 +58,8 @@ int main(int argc, char **argv) {
 
   ::testing::InitGoogleTest(&argc, argv);
   spdlog::set_level(spdlog::level::warn); // Keep test output readable; raise when debugging.
+  // Before anything starts a thread that logs, see nspm_test::log_capture_sink().
+  spdlog::default_logger()->sinks().push_back(nspm_test::log_capture_sink());
   database_manager::init();
   int result = RUN_ALL_TESTS();
 

docker/MQTTManager/tests/mqtt_reconnect_test.cpp

@@ -0,0 +1,89 @@
+#include <gtest/gtest.h>
+#include <mqtt_manager/mqtt_manager.hpp>
+#include <string>
+#include <vector>
+
+// Messages published while disconnected are buffered and sent after reconnecting. Nothing published
+// after reconnecting may overtake them: a retained message that did would leave the server holding
+// the older value.
+class MqttReconnectTest : public ::testing::Test {
+protected:
+  void SetUp() override {
+    // Other tests publish while disconnected too. Send all of that now, so each test starts
+    // disconnected with nothing buffered.
+    MQTT_Manager::test_set_broker_connected(true);
+    MQTT_Manager::test_send_buffered_messages();
+    MQTT_Manager::test_set_broker_connected(false);
+    MQTT_Manager::test_take_sent_messages();
+  }
+
+  void TearDown() override {
+    MQTT_Manager::test_set_broker_connected(false);
+  }
+
+  // Payloads sent to the server on a topic, in order.
+  static std::vector<std::string> sent_to(const std::string &topic) {
+    std::vector<std::string> payloads;
+    for (auto &message : MQTT_Manager::test_take_sent_messages()) {
+      if (message.topic == topic) {
+        payloads.push_back(message.payload);
+      }
+    }
+    return payloads;
+  }
+};
+
+TEST_F(MqttReconnectTest, publishing_after_reconnecting_waits_for_the_buffered_messages) {
+  const std::string topic = "test/mqtt_reconnect/state";
+  MQTT_Manager::publish(topic, "published while disconnected", true);
+
+  // Connected, but not caught up: this is where a newer message used to go out first.
+  MQTT_Manager::test_set_broker_connected(true);
+  EXPECT_FALSE(MQTT_Manager::is_connected());
+  MQTT_Manager::publish(topic, "published after reconnecting", true);
+  EXPECT_TRUE(sent_to(topic).empty());
+
+  MQTT_Manager::test_send_buffered_messages();
+
+  EXPECT_TRUE(MQTT_Manager::is_connected());
+  std::vector<std::string> sent = sent_to(topic);
+  ASSERT_GE(sent.size(), 2);
+  EXPECT_EQ(sent[0], "published while disconnected");
+  EXPECT_EQ(sent.back(), "published after reconnecting");
+}
+
+TEST_F(MqttReconnectTest, each_retained_topic_ends_on_its_latest_value) {
+  const std::string topic = "test/mqtt_reconnect/retained";
+  MQTT_Manager::publish(topic, "old", true);
+  MQTT_Manager::publish(topic, "new", true);
+
+  MQTT_Manager::test_set_broker_connected(true);
+  MQTT_Manager::test_send_buffered_messages();
+
+  std::vector<std::string> sent = sent_to(topic);
+  ASSERT_FALSE(sent.empty());
+  EXPECT_EQ(sent.back(), "new");
+}
+
+TEST_F(MqttReconnectTest, buffered_messages_keep_their_order_and_are_not_replayed_unless_retained) {
+  const std::string topic = "test/mqtt_reconnect/events";
+  MQTT_Manager::publish(topic, "first", false);
+  MQTT_Manager::publish(topic, "second", false);
+
+  MQTT_Manager::test_set_broker_connected(true);
+  MQTT_Manager::test_send_buffered_messages();
+
+  EXPECT_EQ(sent_to(topic), (std::vector<std::string>{"first", "second"}));
+}
+
+TEST_F(MqttReconnectTest, losing_the_connection_buffers_again) {
+  MQTT_Manager::test_set_broker_connected(true);
+  MQTT_Manager::test_send_buffered_messages();
+  EXPECT_TRUE(MQTT_Manager::is_connected());
+
+  MQTT_Manager::test_set_broker_connected(false);
+  EXPECT_FALSE(MQTT_Manager::is_connected());
+
+  MQTT_Manager::test_set_broker_connected(true);
+  EXPECT_FALSE(MQTT_Manager::is_connected());
+}

docker/MQTTManager/tests/test_helpers.hpp

@@ -11,6 +11,7 @@
 #include <nlohmann/json.hpp>
 #include <openhab_manager/openhab_manager.hpp>
 #include <optional>
+#include <spdlog/sinks/dist_sink.h>
 #include <spdlog/sinks/ringbuffer_sink.h>
 #include <spdlog/spdlog.h>
 #include <string>
@@ -211,18 +212,26 @@ inline std::vector<MQTT_Manager::TestPublishedMessage> mqtt_published_to(const s
   return messages;
 }
 
+// The sink that ScopedErrorLog captures through. main() adds it to the default logger once, before any
+// test runs. A logger's own sink list must not change while other threads log through it, and Room
+// threads log for the rest of the run, so captures are added to and removed from this dist_sink
+// instead, which locks its list.
+inline std::shared_ptr<spdlog::sinks::dist_sink_mt> log_capture_sink() {
+  static std::shared_ptr<spdlog::sinks::dist_sink_mt> sink = std::make_shared<spdlog::sinks::dist_sink_mt>();
+  return sink;
+}
+
 // Records the errors (or, given a lower level, also the less severe messages) logged while it is in scope.
 class ScopedErrorLog {
 public:
   explicit ScopedErrorLog(spdlog::level::level_enum level = spdlog::level::err) : _sink(std::make_shared<spdlog::sinks::ringbuffer_sink_mt>(1000)) {
     _sink->set_level(level);
     _sink->set_pattern("%v");
-    spdlog::default_logger()->sinks().push_back(_sink);
+    log_capture_sink()->add_sink(_sink);
   }
 
   ~ScopedErrorLog() {
-    auto &sinks = spdlog::default_logger()->sinks();
-    sinks.erase(std::remove(sinks.begin(), sinks.end(), _sink), sinks.end());
+    log_capture_sink()->remove_sink(_sink);
   }
 
   ScopedErrorLog(const ScopedErrorLog &) = delete;