From f7ed2f176d604a2326b4847136a0b94209bc6dd4 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 8 Aug 2026 19:12:10 -0700 Subject: [PATCH] feat(mqtt): report slot connection health so an outage is visible A slot that cannot establish TLS reported nothing. The per-slot counters track publish successes and failures, and a slot that never connects never attempts a publish -- so its ok count freezes and its error count stays at zero for the entire outage. Observed on hardware: a slot sat dead for over five hours while `sN=` showed 0 errors and the published status payload carried no indication at all. Two surfaces gain the missing signal: - `get mqtt.stats` gains `down=` (omitted when zero, like `filt=`) and marks each disconnected slot with a trailing `!`. The marker goes after the counts so existing "s(\d)=(\d+)/(\d+)" parsers keep matching. - The published status payload gains `mqtt_slots_up`, `mqtt_slots_total`, `heap_largest`, and `mqtt_outage_secs`. These ride on every healthy slot's status message, because the slot that is down cannot report itself. `mqtt_slots_up < mqtt_slots_total` is the condition to alert on. `mqtt_outage_secs` is detail: it is present only when the outage has a known start, so its absence does not imply health -- a slot that has never connected since boot has no outage start time. `heap_largest` is included because free heap alone does not predict a failed connection. A slot needs one contiguous 16,384-byte block for its TLS record buffers; measured immediately before an attempt this value predicted the outcome in every observed case, while total free heap was 85 KB throughout a multi-hour outage. A low value with every slot up is normal rather than a fault -- established sessions already hold their buffers. Publish-error semantics are unchanged: connect failures are not folded into them. Adds host coverage for the new fields, for the outage key's presence rule, and for worst-case payload size. That last one matters because serializeComplete() returns 0 and clears the buffer on overflow and the bridge publishes only a positive length, so exceeding the 768-byte status buffer would silently stop status publication -- the same blindness this change exists to remove. --- src/helpers/MQTTMessageBuilder.cpp | 5 +- src/helpers/MQTTMessageBuilder.h | 4 +- src/helpers/MQTTPayloadBuilder.cpp | 18 ++++- src/helpers/MQTTPayloadBuilder.h | 23 +++++- src/helpers/bridges/MQTTBridge.cpp | 60 +++++++++++++- src/helpers/bridges/MQTTBridge.h | 5 ++ .../test_mqtt_payload_builder.cpp | 79 +++++++++++++++++++ 7 files changed, 184 insertions(+), 10 deletions(-) diff --git a/src/helpers/MQTTMessageBuilder.cpp b/src/helpers/MQTTMessageBuilder.cpp index afb53ef0..fcce321e 100644 --- a/src/helpers/MQTTMessageBuilder.cpp +++ b/src/helpers/MQTTMessageBuilder.cpp @@ -57,13 +57,14 @@ int MQTTMessageBuilder::buildStatusMessage( int internal_heap, int packets_sent, int packets_received, - const char* repeat + const char* repeat, + const MQTTConnHealth& conn_health ) { return MQTTPayloadBuilder::buildStatusMessage( doc, origin, origin_id, model, firmware_version, radio, client_version, status, timestamp, buffer, buffer_size, battery_mv, uptime_secs, errors, queue_len, noise_floor, tx_air_secs, rx_air_secs, recv_errors, internal_heap, - packets_sent, packets_received, repeat); + packets_sent, packets_received, repeat, conn_health); } int MQTTMessageBuilder::buildPacketMessage( diff --git a/src/helpers/MQTTMessageBuilder.h b/src/helpers/MQTTMessageBuilder.h index 91a47b6e..b3d82032 100644 --- a/src/helpers/MQTTMessageBuilder.h +++ b/src/helpers/MQTTMessageBuilder.h @@ -68,6 +68,7 @@ public: * @param recv_errors Radio receive/CRC errors (optional, -1 to omit) * @param internal_heap Internal heap free bytes (optional, -1 to omit) * @param repeat Repeat/forwarding status ("on" or "off"); nullptr omits the field + * @param conn_health MQTT slot connection health; defaults omit every field * @return Length of JSON string, or 0 on error */ static int buildStatusMessage( @@ -93,7 +94,8 @@ public: int internal_heap = -1, int packets_sent = -1, int packets_received = -1, - const char* repeat = nullptr + const char* repeat = nullptr, + const MQTTConnHealth& conn_health = MQTTConnHealth() ); /** diff --git a/src/helpers/MQTTPayloadBuilder.cpp b/src/helpers/MQTTPayloadBuilder.cpp index 47a7b2e8..536bb5e7 100644 --- a/src/helpers/MQTTPayloadBuilder.cpp +++ b/src/helpers/MQTTPayloadBuilder.cpp @@ -45,7 +45,8 @@ int MQTTPayloadBuilder::buildStatusMessage( int internal_heap, int packets_sent, int packets_received, - const char* repeat + const char* repeat, + const MQTTConnHealth& conn_health ) { doc.clear(); JsonObject root = doc.to(); @@ -62,9 +63,12 @@ int MQTTPayloadBuilder::buildStatusMessage( root["repeat"] = repeat; } + const bool has_conn_health = conn_health.slots_up >= 0 || conn_health.slots_total >= 0 || + conn_health.worst_outage_secs >= 0 || conn_health.heap_largest >= 0; + if (battery_mv >= 0 || uptime_secs >= 0 || errors >= 0 || queue_len >= 0 || noise_floor > -999 || tx_air_secs >= 0 || rx_air_secs >= 0 || recv_errors >= 0 || - internal_heap >= 0 || packets_sent >= 0 || packets_received >= 0) { + internal_heap >= 0 || packets_sent >= 0 || packets_received >= 0 || has_conn_health) { JsonObject stats = root["stats"].to(); if (battery_mv >= 0) stats["battery_mv"] = battery_mv; @@ -78,6 +82,16 @@ int MQTTPayloadBuilder::buildStatusMessage( if (rx_air_secs >= 0) stats["rx_air_secs"] = rx_air_secs; if (recv_errors >= 0) stats["recv_errors"] = recv_errors; if (internal_heap >= 0) stats["internal_heap"] = internal_heap; + if (conn_health.heap_largest >= 0) stats["heap_largest"] = conn_health.heap_largest; + if (conn_health.slots_up >= 0) stats["mqtt_slots_up"] = conn_health.slots_up; + if (conn_health.slots_total >= 0) stats["mqtt_slots_total"] = conn_health.slots_total; + // Emitted only during an outage of known duration, so the steady-state payload + // does not grow. Presence implies a slot is down, but absence does NOT imply + // health: a slot that has never connected since boot has no outage start time. + // `mqtt_slots_up < mqtt_slots_total` is the signal to alert on; this is detail. + if (conn_health.worst_outage_secs >= 0) { + stats["mqtt_outage_secs"] = conn_health.worst_outage_secs; + } } return serializeComplete(root, buffer, buffer_size); diff --git a/src/helpers/MQTTPayloadBuilder.h b/src/helpers/MQTTPayloadBuilder.h index ee2a7cc9..b675f397 100644 --- a/src/helpers/MQTTPayloadBuilder.h +++ b/src/helpers/MQTTPayloadBuilder.h @@ -4,6 +4,26 @@ #include #include +// Connection health of the MQTT slots themselves, reported inside a status +// publication. A slot that cannot reach its broker cannot report its own outage, +// so this travels on every *healthy* slot's status message and is how a fleet +// learns that a sibling slot is down. Publish-success/error counters cannot show +// this: a slot that never connects never attempts a publish, so its error count +// stays at zero for the entire outage. +// +// `heap_largest` is the leading indicator. Observed on non-PSRAM hardware: a slot +// needs one contiguous 16,384-byte block for its TLS record buffers, and measured +// immediately before an attempt this value predicted the outcome every time. Free +// heap alone does not — an 85 KB-free board sat unable to connect for hours. +// A low value with all slots up is normal, not a fault: established sessions +// already hold their buffers. It means only that the next slot to ask will fail. +struct MQTTConnHealth { + int slots_up = -1; // -1 = not supplied, field omitted + int slots_total = -1; // enabled slots, whether or not they connect + int worst_outage_secs = -1; // longest current outage; -1 = nothing is down + int heap_largest = -1; // largest free internal block, bytes +}; + // Mesh-independent JSON serialization core for MQTT publication payloads. // MQTTMessageBuilder keeps the firmware-facing API and delegates these three // deterministic contracts here so they can be exercised by native tests. @@ -37,7 +57,8 @@ public: int internal_heap = -1, int packets_sent = -1, int packets_received = -1, - const char* repeat = nullptr + const char* repeat = nullptr, + const MQTTConnHealth& conn_health = MQTTConnHealth() ); static int buildPacketMessage( diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 29edc89e..bde48149 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -373,12 +373,26 @@ void MQTTBridge::formatMqttStatsReply(char* buf, size_t bufsize) { if (b->_filtered_packets > 0) { replyAppendf(buf, bufsize, &pos, " filt=%lu", b->_filtered_packets); } + // down=: enabled slots that are not connected. Without this the reply cannot + // show an outage at all — sN= counts publishes, and a slot that never connects + // never publishes, so its ok freezes and its err stays 0 for the whole outage. + // Omitted while zero, like filt=, so a healthy node's reply keeps its length. + int down = 0; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (b->_slots[i].enabled && !b->_slots[i].connected) down++; + } + if (down > 0) { + replyAppendf(buf, bufsize, &pos, " down=%d", down); + } replyAppendf(buf, bufsize, &pos, " |"); for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { if (!b->_slots[i].enabled || !b->_slots[i].client) continue; - replyAppendf(buf, bufsize, &pos, " s%d=%lu/%lu", i + 1, + // Trailing '!' marks a disconnected slot. Appended after the counts so the + // established "s(\d)=(\d+)/(\d+)" parsers keep matching unchanged. + replyAppendf(buf, bufsize, &pos, " s%d=%lu/%lu%s", i + 1, b->_slots[i].client->getPublishOk(), - b->_slots[i].client->getPublishErr()); + b->_slots[i].client->getPublishErr(), + b->_slots[i].connected ? "" : "!"); } } @@ -3135,7 +3149,8 @@ void MQTTBridge::publishStatusToSlot(int index) { battery_mv, uptime_secs, errors, _queue_count, noise_floor, tx_air_secs, rx_air_secs, recv_errors, internal_heap_free, packets_sent, packets_received, - _prefs->disable_fwd ? "off" : "on" + _prefs->disable_fwd ? "off" : "on", + collectConnHealth() ); if (len > 0) { @@ -3151,6 +3166,42 @@ void MQTTBridge::publishStatusToSlot(int index) { } } +MQTTConnHealth MQTTBridge::collectConnHealth() const { + MQTTConnHealth health; + health.slots_up = 0; + health.slots_total = 0; + + const unsigned long now = millis(); + unsigned long worst_outage_ms = 0; + bool any_outage_timed = false; + + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + const MQTTSlot& slot = _slots[i]; + if (!slot.enabled) continue; // "none" slots are not an outage, they are unconfigured + health.slots_total++; + if (slot.connected) { + health.slots_up++; + continue; + } + // A slot that has never connected since boot has no outage start time. It still + // counts against slots_up, which is why that pair — not this duration — is the + // signal to alert on. + if (slot.current_outage_started_ms != 0) { + const unsigned long outage_ms = now - slot.current_outage_started_ms; // wrap-safe + if (!any_outage_timed || outage_ms > worst_outage_ms) { + worst_outage_ms = outage_ms; + any_outage_timed = true; + } + } + } + + if (any_outage_timed) { + health.worst_outage_secs = static_cast(worst_outage_ms / 1000UL); + } + health.heap_largest = static_cast(heap_caps_get_largest_free_block(MALLOC_CAP_INTERNAL)); + return health; +} + void MQTTBridge::updateCachedConnectionStatus() { bool any_connected = false; for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { @@ -3993,7 +4044,8 @@ bool MQTTBridge::publishStatus() { battery_mv, uptime_secs, errors, _queue_count, noise_floor, tx_air_secs, rx_air_secs, recv_errors, internal_heap_free, packets_sent, packets_received, - _prefs->disable_fwd ? "off" : "on" + _prefs->disable_fwd ? "off" : "on", + collectConnHealth() ); if (len > 0) { diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 1126da0b..ca7e061a 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -9,6 +9,7 @@ #include #include "helpers/JWTHelper.h" #include "helpers/MQTTPacketFilter.h" +#include "helpers/MQTTPayloadBuilder.h" // MQTTConnHealth #include "helpers/MQTTPresets.h" #include "helpers/MQTTPresetPolicy.h" #include "helpers/MQTTLifecycle.h" @@ -542,6 +543,10 @@ private: bool publishToSlot(int index, const char* topic, const char* payload, size_t payload_len, bool retained = false, uint8_t qos = 0); bool publishToAllSlots(const char* topic, const char* payload, size_t payload_len, bool retained = false, uint8_t qos = 0); void publishStatusToSlot(int index); + // Snapshot of slot connection health for the status payload. A slot that cannot + // reach its broker cannot report its own outage, so every healthy slot carries + // this on its status publication. + MQTTConnHealth collectConnHealth() const; void updateCachedConnectionStatus(); void processPacketQueue(); diff --git a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp index 181f0a86..e682f8b4 100644 --- a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp +++ b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp @@ -93,6 +93,85 @@ TEST(MQTTPayloadBuilder, StatusIncludesRepeatAndEveryRequestedStatistic) { EXPECT_EQ(47, stats["packets_received"].as()); } +TEST(MQTTPayloadBuilder, StatusReportsConnectionHealthWhenSupplied) { + JsonDocument scratch; + char buffer[1024]; + MQTTConnHealth health; + health.slots_up = 2; + health.slots_total = 3; + health.worst_outage_secs = 18432; + health.heap_largest = 16116; + + int len = MQTTPayloadBuilder::buildStatusMessage( + scratch, "node", "id", "model", "firmware", "radio", "client", "online", + kTimestamp, buffer, sizeof(buffer), -1, -1, -1, -1, -999, + -1, -1, -1, -1, -1, -1, nullptr, health); + + ASSERT_GT(len, 0); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + JsonObject stats = parsed["stats"].as(); + ASSERT_FALSE(stats.isNull()); + EXPECT_EQ(2, stats["mqtt_slots_up"].as()); + EXPECT_EQ(3, stats["mqtt_slots_total"].as()); + EXPECT_EQ(18432, stats["mqtt_outage_secs"].as()); + EXPECT_EQ(16116, stats["heap_largest"].as()); +} + +// The outage key doubles as the "something is down" flag, so its absence has to be +// meaningful: a fully healthy board must not emit it. +TEST(MQTTPayloadBuilder, StatusOmitsOutageKeyWhenEverySlotIsUp) { + JsonDocument scratch; + char buffer[768]; + MQTTConnHealth health; + health.slots_up = 3; + health.slots_total = 3; + health.heap_largest = 54260; // worst_outage_secs left at -1 + + int len = MQTTPayloadBuilder::buildStatusMessage( + scratch, "node", "id", "model", "firmware", "radio", "client", "online", + kTimestamp, buffer, sizeof(buffer), -1, -1, -1, -1, -999, + -1, -1, -1, -1, -1, -1, nullptr, health); + + ASSERT_GT(len, 0); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + JsonObject stats = parsed["stats"].as(); + ASSERT_FALSE(stats.isNull()); + EXPECT_EQ(3, stats["mqtt_slots_up"].as()); + EXPECT_FALSE(stats["mqtt_outage_secs"].is()); +} + +// serializeComplete() returns 0 and clears the buffer when the payload does not fit, +// and MQTTBridge publishes only a positive length — so overflowing STATUS_JSON_BUFFER_SIZE +// silently stops status publication altogether. That is the same class of blindness this +// field set exists to fix, so the worst realistic payload must fit with margin. +TEST(MQTTPayloadBuilder, WorstCaseStatusWithConnectionHealthFitsProductionBuffer) { + constexpr size_t kStatusJsonBufferSize = 768; // MQTTBridge::STATUS_JSON_BUFFER_SIZE + JsonDocument scratch; + char buffer[kStatusJsonBufferSize]; + MQTTConnHealth health; + health.slots_up = 0; + health.slots_total = 5; // PSRAM slot cap, the widest configuration + health.worst_outage_secs = 999999; + health.heap_largest = 4194304; + + int len = MQTTPayloadBuilder::buildStatusMessage( + scratch, + "A Repeater With A Deliberately Long Node Name", + "0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF", // 64-char device id + "Heltec Wireless Tracker V1.1", "v1.17.0-observer-dev.12", + "915.000000,62.5,7,5", "MeshCore 1.17.0", "online", kTimestamp, + buffer, sizeof(buffer), + 4200, 8640000, 65535, 6, -128, 999999, 999999, 65535, 4194304, 999999, 999999, + "off", health); + + ASSERT_GT(len, 0) << "worst-case status payload overflowed the production buffer; " + "status publication would stop silently"; + EXPECT_LT(static_cast(len), kStatusJsonBufferSize - 32) + << "less than 32 bytes of headroom left in the status buffer"; +} + TEST(MQTTPayloadBuilder, StatusOmissionSentinelsRemainOmitted) { JsonDocument scratch; char buffer[768];