diff --git a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp index a81ab4b9..5d697f90 100644 --- a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp +++ b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.cpp @@ -119,6 +119,15 @@ 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 { @@ -126,6 +135,44 @@ bool MQTT_Manager::is_connected() { } } +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 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 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 retained_messages; + { + std::lock_guard 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 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(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 retained_messages; - { - std::lock_guard 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 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); } diff --git a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp b/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp index 98425a2d..19b84b0b 100644 --- a/docker/MQTTManager/include/mqtt_manager/mqtt_manager.hpp +++ b/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 @@ -129,12 +134,35 @@ class MQTT_Manager { std::lock_guard 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 test_take_sent_messages() { + std::lock_guard 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 _test_published_messages; + static inline std::vector _test_sent_messages; + static inline std::atomic _test_broker_connected = false; #endif static inline boost::lockfree::spsc_queue> _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 _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> _mqtt_callbacks; static inline std::unordered_map _subscribed_topics; diff --git a/docker/MQTTManager/tests/main.cpp b/docker/MQTTManager/tests/main.cpp index 636e56e0..c569b34f 100644 --- a/docker/MQTTManager/tests/main.cpp +++ b/docker/MQTTManager/tests/main.cpp @@ -2,6 +2,7 @@ #include #include #include +#include "test_helpers.hpp" #include #include #include @@ -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(); diff --git a/docker/MQTTManager/tests/mqtt_reconnect_test.cpp b/docker/MQTTManager/tests/mqtt_reconnect_test.cpp new file mode 100644 index 00000000..34628e98 --- /dev/null +++ b/docker/MQTTManager/tests/mqtt_reconnect_test.cpp @@ -0,0 +1,89 @@ +#include +#include +#include +#include + +// 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 sent_to(const std::string &topic) { + std::vector 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 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 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{"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()); +} diff --git a/docker/MQTTManager/tests/test_helpers.hpp b/docker/MQTTManager/tests/test_helpers.hpp index 5930b3c3..1a1759a0 100644 --- a/docker/MQTTManager/tests/test_helpers.hpp +++ b/docker/MQTTManager/tests/test_helpers.hpp @@ -11,6 +11,7 @@ #include #include #include +#include #include #include #include @@ -211,18 +212,26 @@ inline std::vector 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 log_capture_sink() { + static std::shared_ptr sink = std::make_shared(); + 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(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;