From 270e8295823400e749613c534762b408ed435ef8 Mon Sep 17 00:00:00 2001 From: James Mulcahy Date: Sat, 17 Oct 2026 13:02:59 -0700 Subject: [PATCH 1/2] MQTTManager: don't let new messages overtake ones buffered while disconnected Messages published while disconnected from MQTT are buffered and sent after reconnecting. But publish() went back to sending live as soon as the client reported connected, while the reconnect still had to replay the retained values and subscribe to every topic (5 ms each) before it sent the buffer. Anything published in that window went out first and was then overwritten by an older buffered message. On a real install, a media player's state published before the first connection (only its ID and name, as Home Assistant had not answered yet) was buffered, the real state was published live half a second later, and the buffered placeholder then replaced it on the broker. The manager saw no change to send after that, so panels showed an empty media player for hours. The retained replay had the same problem: it published from a copy taken earlier, through publish(), so it could send an older value after a newer one and write the older value back into the retain buffer. is_connected() now also requires that the buffered messages have been sent. Until then publish() and clear_retain() buffer, behind what is already waiting. After connecting, the buffer is sent in order, then the latest value of every retained topic, all under the client lock, and only then are topics subscribed to again. Co-Authored-By: Claude Opus 5.5 --- .../include/mqtt_manager/mqtt_manager.cpp | 81 +++++++++++------ .../include/mqtt_manager/mqtt_manager.hpp | 44 +++++++++ .../MQTTManager/tests/mqtt_reconnect_test.cpp | 89 +++++++++++++++++++ 3 files changed, 189 insertions(+), 25 deletions(-) create mode 100644 docker/MQTTManager/tests/mqtt_reconnect_test.cpp 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/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()); +} From 0451a6cd39af64dd4c4abfe1822a0df73dfd194c Mon Sep 17 00:00:00 2001 From: James Mulcahy Date: Sat, 17 Oct 2026 13:13:29 -0700 Subject: [PATCH 2/2] Tests: stop changing the logger's sinks while Room threads log ScopedErrorLog pushed its sink onto, and erased it from, the default logger's sink vector. spdlog does not lock that vector, and every Room runs a thread that logs for the rest of the run, so a capture starting or ending while one of those threads logged could reallocate the vector under it. That crashed a CI run in spdlog::logger::sink_it_, called from Room::_update_room_state(), between tests. main() now adds a single dist_sink to the default logger before any test runs, and ScopedErrorLog adds and removes its sink through that, which locks its own list. Co-Authored-By: Claude Opus 5.5 --- docker/MQTTManager/tests/main.cpp | 3 +++ docker/MQTTManager/tests/test_helpers.hpp | 15 ++++++++++++--- 2 files changed, 15 insertions(+), 3 deletions(-) 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/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;