From f7ed2f176d604a2326b4847136a0b94209bc6dd4 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 8 Aug 2026 19:12:10 -0700 Subject: [PATCH 1/7] 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 afb53ef0dd..fcce321ea6 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 91a47b6eb0..b3d8203295 100644 --- a/src/helpers/MQTTMessageBuilder.h +++ b/src/helpers/MQTTMessageBuilder.h @@ -68,6 +68,7 @@ class MQTTMessageBuilder { * @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 @@ class MQTTMessageBuilder { 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 47a7b2e8b9..536bb5e7f2 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 ee2a7cc997..b675f397cb 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 @@ class MQTTPayloadBuilder { 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 29edc89e75..bde481490a 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 1126da0bcb..ca7e061abe 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 @@ class MQTTBridge : public BridgeBase { 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 181f0a86cd..e682f8b408 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]; From 383339ecccf611bd83458b189dfd4ff20d5d8fc1 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 26 Sep 2026 11:09:21 -0700 Subject: [PATCH 2/7] fix(mqtt): size the status buffer for real field maxima The worst-case test added with the connection-health fields did not model production maxima -- it used a 28-byte model and a 24-byte firmware string, while MQTTBridge permits _origin[32], _device_id[65], _board_model[64], _firmware_version[64] and derives a 63-byte client version. isValidName() rejects [ ] / \ : , ? * but permits '"', so a 31-character node name of quotes is legal and every quote escapes to two bytes. Measured at those real maxima: without connection-health fields 762 bytes with connection-health fields 851 bytes buffer 768 bytes So the 768-byte buffer already had only six bytes of headroom before this work, and the four new fields overflow it. Overflow is silent: serializeComplete() returns 0 and clears the buffer, and the bridge publishes only a positive length, so a device would simply stop publishing status with no error anywhere -- the same blindness the connection-health fields exist to remove. Raises STATUS_JSON_BUFFER_SIZE to 1024 and corrects the test to use the actual maxima. Status serializes into the shared 2048-byte publish scratch buffer, so this costs no extra RAM; only the PSRAM-allocation-failed stack fallback grows by 256 bytes. A static_assert keeps the status ceiling within the shared buffer. Found by adversarial review of the preceding commit. --- src/helpers/bridges/MQTTBridge.h | 12 ++++-- .../test_mqtt_payload_builder.cpp | 37 +++++++++++++------ 2 files changed, 34 insertions(+), 15 deletions(-) diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index ca7e061abe..b755ce2f3f 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -366,9 +366,15 @@ class MQTTBridge : public BridgeBase { // internal heap. On non-PSRAM: inline in the class object so the allocation doesn't // interleave with large TLS buffers at startup. static const size_t PUBLISH_JSON_BUFFER_SIZE = 2048; - // Status keeps its own smaller ceiling: raising it would change which oversized - // status documents get published instead of dropped. - static const size_t STATUS_JSON_BUFFER_SIZE = 768; + // Status keeps its own smaller ceiling: raising it changes which oversized status + // documents get published instead of dropped. 768 left only 6 bytes spare at the real + // field maxima (_origin[32] of quotes, which isValidName permits and JSON doubles, plus + // _device_id[65], _board_model[64], _firmware_version[64] and a 63-byte client version): + // measured 762/768 before the connection-health fields existed. Overflow is silent — + // serializeComplete() returns 0 and the bridge publishes only len>0, so status stops + // entirely. Raised to 1024, which costs no extra RAM here because status serializes into + // the shared scratch buffer. See the worst-case test in test_mqtt_payload_builder. + static const size_t STATUS_JSON_BUFFER_SIZE = 1024; static_assert(STATUS_JSON_BUFFER_SIZE <= PUBLISH_JSON_BUFFER_SIZE, "status payloads serialize into the shared publish buffer"); #if defined(BOARD_HAS_PSRAM) 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 e682f8b408..b0c2d438f0 100644 --- a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp +++ b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp @@ -146,30 +146,43 @@ TEST(MQTTPayloadBuilder, StatusOmitsOutageKeyWhenEverySlotIsUp) { // 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. +// Field widths are the real MQTTBridge maxima, not plausible-looking strings: +// _origin[32] _device_id[65] _firmware_version[64] _board_model[64] +// and client_version is snprintf("meshcore/%s", _firmware_version) into char[64]. +// isValidName() rejects [ ] / \ : , ? * but ALLOWS '"', so a node name of 31 quotes +// is legal and every one escapes to two bytes — that is the true worst case for the +// only free-text field an operator controls. TEST(MQTTPayloadBuilder, WorstCaseStatusWithConnectionHealthFitsProductionBuffer) { - constexpr size_t kStatusJsonBufferSize = 768; // MQTTBridge::STATUS_JSON_BUFFER_SIZE + constexpr size_t kStatusJsonBufferSize = 1024; // MQTTBridge::STATUS_JSON_BUFFER_SIZE + const std::string origin(31, '"'); // _origin[32], worst-case escaping + const std::string origin_id(64, 'A'); // _device_id[65] + const std::string model(63, 'M'); // _board_model[64] + const std::string firmware(63, 'F'); // _firmware_version[64] + const std::string client_version(63, 'C'); // char client_version[64] + const std::string radio("9999.999999,1000.0,12,8"); + 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; + health.worst_outage_secs = 4294967; // millis() rollover / 1000, the widest value + health.heap_largest = 8388608; 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, + scratch, origin.c_str(), origin_id.c_str(), model.c_str(), firmware.c_str(), + radio.c_str(), client_version.c_str(), "online", kTimestamp, buffer, sizeof(buffer), - 4200, 8640000, 65535, 6, -128, 999999, 999999, 65535, 4194304, 999999, 999999, + 4200, 2147483, 2147483647, 6, -128, 2147483, 2147483, 2147483647, + 8388608, 2147483647, 2147483647, "off", health); - ASSERT_GT(len, 0) << "worst-case status payload overflowed the production buffer; " - "status publication would stop silently"; + ASSERT_GT(len, 0) << "worst-case status payload overflowed the " << kStatusJsonBufferSize + << "-byte production buffer; serializeComplete() returns 0 and the " + "bridge publishes only len>0, so ALL status publication would " + "stop silently"; EXPECT_LT(static_cast(len), kStatusJsonBufferSize - 32) - << "less than 32 bytes of headroom left in the status buffer"; + << "less than 32 bytes of headroom left in the status buffer (len=" << len << ")"; } TEST(MQTTPayloadBuilder, StatusOmissionSentinelsRemainOmitted) { From 7004b81c0c4773b642663cd7218bf96456d19283 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 26 Sep 2026 11:10:13 -0700 Subject: [PATCH 3/7] feat(mqtt): count connect failures and report tripped breakers Two gaps remained after the first connection-health commit. A slot retrying into a heap that cannot satisfy its TLS allocation moved no counter at all. disconnect_count rose, but it also rises for dropped live sessions, so it could not separate "actively failing to connect" from "was connected, dropped once". Adds a per-slot connect_failures, incremented in onDisconnect when the event arrives while the client is Starting -- meaning it ended an attempt that never reached onConnect. Late events from a stopped or quarantined client, and soft-disconnects of a live session, are in other states and are not miscounted. A slot parked by the circuit breaker also looked identical to one still retrying, despite being materially worse: the breaker only probes every 30 minutes, so recovery is far slower and an operator may need to intervene. Surfaces both: - `mqttN.diag` gains `cf:` beside the existing `dc:`. - the status payload gains `mqtt_connect_failures` (summed across slots) and `mqtt_slots_breaker` (count of tripped slots), each emitted only when non-zero so a healthy board's payload does not grow. The counter is what distinguishes a retrying slot from a quiet one: while a slot is down its publish counters cannot move by definition, so this is the only value that shows the device is still trying. Both suggested by adversarial review. --- src/helpers/MQTTPayloadBuilder.cpp | 9 +++- src/helpers/MQTTPayloadBuilder.h | 7 ++++ src/helpers/bridges/MQTTBridge.cpp | 20 +++++++++ src/helpers/bridges/MQTTBridge.h | 4 ++ .../test_mqtt_payload_builder.cpp | 42 +++++++++++++++++++ 5 files changed, 81 insertions(+), 1 deletion(-) diff --git a/src/helpers/MQTTPayloadBuilder.cpp b/src/helpers/MQTTPayloadBuilder.cpp index 536bb5e7f2..9811b98f76 100644 --- a/src/helpers/MQTTPayloadBuilder.cpp +++ b/src/helpers/MQTTPayloadBuilder.cpp @@ -64,7 +64,8 @@ int MQTTPayloadBuilder::buildStatusMessage( } 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; + conn_health.worst_outage_secs >= 0 || conn_health.heap_largest >= 0 || + conn_health.connect_failures >= 0 || conn_health.slots_breaker >= 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 || @@ -92,6 +93,12 @@ int MQTTPayloadBuilder::buildStatusMessage( if (conn_health.worst_outage_secs >= 0) { stats["mqtt_outage_secs"] = conn_health.worst_outage_secs; } + if (conn_health.connect_failures > 0) { + stats["mqtt_connect_failures"] = conn_health.connect_failures; + } + if (conn_health.slots_breaker > 0) { + stats["mqtt_slots_breaker"] = conn_health.slots_breaker; + } } return serializeComplete(root, buffer, buffer_size); diff --git a/src/helpers/MQTTPayloadBuilder.h b/src/helpers/MQTTPayloadBuilder.h index b675f397cb..42dd3ff868 100644 --- a/src/helpers/MQTTPayloadBuilder.h +++ b/src/helpers/MQTTPayloadBuilder.h @@ -22,6 +22,13 @@ struct MQTTConnHealth { 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 + // Attempts that ended without reaching onConnect, summed across slots. A slot + // stuck retrying moves this while its publish counters stay frozen, which is + // what separates "actively failing" from "idle and quiet". + int connect_failures = -1; + // Slots parked by the circuit breaker. Materially worse than a retrying slot: + // the breaker only probes every 30 minutes, so recovery is far slower. + int slots_breaker = -1; }; // Mesh-independent JSON serialization core for MQTT publication payloads. diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index bde481490a..bcff691726 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -584,6 +584,12 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) replyAppendf(buf, bufsize, &pos, "> mqtt%d: %s", slot_index + 1, state); if (slot.disconnect_count > 0) { replyAppendf(buf, bufsize, &pos, ", dc:%lu", (unsigned long)slot.disconnect_count); + // cf: attempts that never reached onConnect. A slot retrying into a heap that + // cannot satisfy the TLS allocation moves only this counter — dc rises with it, + // but the publish counters cannot move at all while the slot is down. + if (slot.connect_failures > 0) { + replyAppendf(buf, bufsize, &pos, ", cf:%lu", (unsigned long)slot.connect_failures); + } if (slot.first_disconnect_time > 0) { unsigned long first_disc_age_sec = (millis() - slot.first_disconnect_time) / 1000; replyAppendf(buf, bufsize, &pos, ", first_disc:%lus", first_disc_age_sec); @@ -1757,6 +1763,11 @@ bool MQTTBridge::ensureSlotClient(int index) { }); slot.client->onDisconnect([this, index](bool sessionPresent) { MQTT_DEBUG_PRINTLN("MQTT%d disconnected", index + 1); + // An event while an attempt is in flight ended one that never reached onConnect. + // Stopped/quarantined late events and soft-disconnects are not Starting. + if (mqttClientStateHasAttemptInFlight(_slots[index].client_state)) { + _slots[index].connect_failures++; + } // Only a live client's disconnect is news. One arriving for a client we // already stopped (or quarantined) must not resurrect its state. if (clientStateIsLive(_slots[index].client_state)) { @@ -3174,11 +3185,15 @@ MQTTConnHealth MQTTBridge::collectConnHealth() const { const unsigned long now = millis(); unsigned long worst_outage_ms = 0; bool any_outage_timed = false; + uint32_t connect_failures = 0; + int breaker_slots = 0; 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++; + connect_failures += slot.connect_failures; + if (slot.circuit_breaker_tripped) breaker_slots++; if (slot.connected) { health.slots_up++; continue; @@ -3198,6 +3213,11 @@ MQTTConnHealth MQTTBridge::collectConnHealth() const { if (any_outage_timed) { health.worst_outage_secs = static_cast(worst_outage_ms / 1000UL); } + // Clamp: the payload field is int, and these are lifetime counters. + health.connect_failures = connect_failures > static_cast(INT32_MAX) + ? INT32_MAX + : static_cast(connect_failures); + health.slots_breaker = breaker_slots; health.heap_largest = static_cast(heap_caps_get_largest_free_block(MALLOC_CAP_INTERNAL)); return health; } diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index b755ce2f3f..bbf04472c7 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -183,6 +183,10 @@ class MQTTBridge : public BridgeBase { uint8_t last_connack_code; unsigned long last_error_time; // millis() of last error uint32_t disconnect_count; // Number of disconnect callbacks since boot + // Attempts that ended without ever reaching onConnect. Distinct from + // disconnect_count (which also counts dropped live sessions) and from the + // publish error counter, which cannot move at all while a slot is down. + uint32_t connect_failures; unsigned long first_disconnect_time; // millis() of first disconnect after boot // Current-outage timer (used by AlertReporter to fire faults after a sustained 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 b0c2d438f0..f539ee8ddd 100644 --- a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp +++ b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp @@ -118,6 +118,46 @@ TEST(MQTTPayloadBuilder, StatusReportsConnectionHealthWhenSupplied) { EXPECT_EQ(16116, stats["heap_largest"].as()); } +TEST(MQTTPayloadBuilder, StatusReportsConnectFailuresAndBreakerOnlyWhenNonZero) { + JsonDocument scratch; + char buffer[1024]; + MQTTConnHealth health; + health.slots_up = 1; + health.slots_total = 3; + health.connect_failures = 57; + health.slots_breaker = 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(); + EXPECT_EQ(57, stats["mqtt_connect_failures"].as()); + EXPECT_EQ(1, stats["mqtt_slots_breaker"].as()); + + // A healthy board must not carry either key: zero connect failures and zero + // tripped breakers are the normal case, and the steady-state payload is + // already close to the buffer bound. + MQTTConnHealth healthy; + healthy.slots_up = 3; + healthy.slots_total = 3; + healthy.connect_failures = 0; + healthy.slots_breaker = 0; + ASSERT_GT(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, healthy), 0); + JsonDocument healthy_parsed; + ASSERT_FALSE(deserializeJson(healthy_parsed, buffer)); + JsonObject healthy_stats = healthy_parsed["stats"].as(); + EXPECT_FALSE(healthy_stats["mqtt_connect_failures"].is()); + EXPECT_FALSE(healthy_stats["mqtt_slots_breaker"].is()); +} + // 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) { @@ -168,6 +208,8 @@ TEST(MQTTPayloadBuilder, WorstCaseStatusWithConnectionHealthFitsProductionBuffer health.slots_total = 5; // PSRAM slot cap, the widest configuration health.worst_outage_secs = 4294967; // millis() rollover / 1000, the widest value health.heap_largest = 8388608; + health.connect_failures = 2147483647; + health.slots_breaker = 5; int len = MQTTPayloadBuilder::buildStatusMessage( scratch, origin.c_str(), origin_id.c_str(), model.c_str(), firmware.c_str(), From 73098a44b6d958a54851334cbc02bf7b1f85df8a Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 26 Sep 2026 11:53:49 -0700 Subject: [PATCH 4/7] fix(mqtt): count connect failures from a flag set before the attempt The Starting-state gate raced the event task: client_state becomes Starting only after connect()/reconnect() returns, so a fast failure could deliver DISCONNECTED first and go uncounted, while a deliberate stop of an attempt in flight was counted as a failure. A per-slot flag is now raised before the call and cleared by onConnect, a failed start, or stopSlotClient(). Also stops zero-valued connect_failures/slots_breaker from creating an empty stats object. --- src/helpers/MQTTPayloadBuilder.cpp | 2 +- src/helpers/bridges/MQTTBridge.cpp | 13 ++++++++++--- src/helpers/bridges/MQTTBridge.h | 5 +++++ .../test_mqtt_payload_builder.cpp | 17 +++++++++++++++++ 4 files changed, 33 insertions(+), 4 deletions(-) diff --git a/src/helpers/MQTTPayloadBuilder.cpp b/src/helpers/MQTTPayloadBuilder.cpp index 9811b98f76..8cbb03f4ce 100644 --- a/src/helpers/MQTTPayloadBuilder.cpp +++ b/src/helpers/MQTTPayloadBuilder.cpp @@ -65,7 +65,7 @@ int MQTTPayloadBuilder::buildStatusMessage( 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 || - conn_health.connect_failures >= 0 || conn_health.slots_breaker >= 0; + conn_health.connect_failures > 0 || conn_health.slots_breaker > 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 || diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index bcff691726..134b12e7c4 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -772,6 +772,7 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg _slot_reconfigure_pending[i] = false; _slot_force_jwt_mint[i] = false; _status_publish_pending[i] = false; + _slot_attempt_pending[i] = false; } // Reset CLI-requested forced NTP sync handshake (bridge object is reused across restarts) @@ -1730,6 +1731,7 @@ bool MQTTBridge::ensureSlotClient(int index) { return; } MQTT_DEBUG_PRINTLN("MQTT%d connected", index + 1); + _slot_attempt_pending[index] = false; _slots[index].client_state = ClientState::Connected; _slots[index].connected = true; _slot_force_jwt_mint[index] = false; @@ -1763,9 +1765,10 @@ bool MQTTBridge::ensureSlotClient(int index) { }); slot.client->onDisconnect([this, index](bool sessionPresent) { MQTT_DEBUG_PRINTLN("MQTT%d disconnected", index + 1); - // An event while an attempt is in flight ended one that never reached onConnect. - // Stopped/quarantined late events and soft-disconnects are not Starting. - if (mqttClientStateHasAttemptInFlight(_slots[index].client_state)) { + // Ended an attempt that never reached onConnect. A deliberate stop clears the flag + // first, so cancelling an attempt is not counted as a failure. + if (_slot_attempt_pending[index]) { + _slot_attempt_pending[index] = false; _slots[index].connect_failures++; } // Only a live client's disconnect is news. One arriving for a client we @@ -2411,6 +2414,7 @@ void MQTTBridge::closeLiveClientForLinkTransition(int index) { // the rest of the boot. void MQTTBridge::stopSlotClient(int index) { MQTTSlot& slot = _slots[index]; + _slot_attempt_pending[index] = false; const esp_err_t r = slot.client->disconnect(); // What the SDK's results actually mean here (mqtt_client.h documents @@ -2505,6 +2509,7 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { } esp_err_t r; + _slot_attempt_pending[index] = true; if (!slot.client->isStarted()) { MQTT_DEBUG_PRINTLN("MQTT%d start (client was stopped)", index + 1); r = slot.client->connect(); @@ -2517,6 +2522,8 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { // The attempt carries whatever credential is configured right now, so this // is the point at which a freshly minted token becomes the one in use. slot.applied_token_expires_at = slot.token_expires_at; + } else { + _slot_attempt_pending[index] = false; } return r; } diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index bbf04472c7..87b75b050f 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -286,6 +286,11 @@ class MQTTBridge : public BridgeBase { // _slot_reconfigure_pending; a single-byte volatile store/load is atomic. volatile bool _status_publish_pending[RUNTIME_MQTT_SLOTS]; + // Set by the bridge task before connect()/reconnect(), so a failure the event task + // delivers before that call returns is still seen; cleared by the attempt's outcome + // or a deliberate stop. A disconnect that finds it set is a connect failure. + volatile bool _slot_attempt_pending[RUNTIME_MQTT_SLOTS]; + // CLI-requested forced NTP sync, marshalled onto the MQTT task (Core 0). // All NTP I/O must run on Core 0; the CLI thread // (Core 1) sets _ntp_force_requested and blocks in requestForcedNtpSync() 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 f539ee8ddd..e8ae51c871 100644 --- a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp +++ b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp @@ -158,6 +158,23 @@ TEST(MQTTPayloadBuilder, StatusReportsConnectFailuresAndBreakerOnlyWhenNonZero) EXPECT_FALSE(healthy_stats["mqtt_slots_breaker"].is()); } +// Zero counters with nothing else supplied must not create an empty stats object. +TEST(MQTTPayloadBuilder, StatusOmitsStatsWhenOnlyZeroCountersSupplied) { + JsonDocument scratch; + char buffer[768]; + MQTTConnHealth health; + health.connect_failures = 0; + health.slots_breaker = 0; + + ASSERT_GT(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), 0); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + EXPECT_FALSE(parsed["stats"].is()); +} + // 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) { From 6390763c719998ed7000820aaac97a1c87e9a227 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 26 Sep 2026 12:25:53 -0700 Subject: [PATCH 5/7] fix(mqtt): count failed starts and ignore bounced sessions in cf connect()/reconnect() returning an error started no attempt, so no event arrived and the failure went uncounted -- the low-heap case the counter exists for. Counted now in a bridge-task-only start_failures, summed into cf: and mqtt_connect_failures. A renewal or corrected-clock bounce whose softDisconnect() times out gets its old session's DISCONNECTED after the new attempt raised the flag. That session is still marked connected until the handler runs, so the gate now also requires !connected. An ignored late CONNECTED clears the flag too. cf: no longer depends on dc being non-zero. Comments corrected for the PSRAM-fallback stack cost and heap_largest's sampling point; the worst-case test now models 6 slots. --- src/helpers/MQTTPayloadBuilder.h | 2 ++ src/helpers/bridges/MQTTBridge.cpp | 22 +++++++++++-------- src/helpers/bridges/MQTTBridge.h | 11 +++++++--- .../test_mqtt_payload_builder.cpp | 4 ++-- 4 files changed, 25 insertions(+), 14 deletions(-) diff --git a/src/helpers/MQTTPayloadBuilder.h b/src/helpers/MQTTPayloadBuilder.h index 42dd3ff868..7f0cd85b0f 100644 --- a/src/helpers/MQTTPayloadBuilder.h +++ b/src/helpers/MQTTPayloadBuilder.h @@ -17,6 +17,8 @@ // 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. +// Sampled at status-publish time, not at the attempt, so it is a trend, not a +// per-attempt reading. struct MQTTConnHealth { int slots_up = -1; // -1 = not supplied, field omitted int slots_total = -1; // enabled slots, whether or not they connect diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 134b12e7c4..2bee5bb250 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -584,17 +584,18 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) replyAppendf(buf, bufsize, &pos, "> mqtt%d: %s", slot_index + 1, state); if (slot.disconnect_count > 0) { replyAppendf(buf, bufsize, &pos, ", dc:%lu", (unsigned long)slot.disconnect_count); - // cf: attempts that never reached onConnect. A slot retrying into a heap that - // cannot satisfy the TLS allocation moves only this counter — dc rises with it, - // but the publish counters cannot move at all while the slot is down. - if (slot.connect_failures > 0) { - replyAppendf(buf, bufsize, &pos, ", cf:%lu", (unsigned long)slot.connect_failures); - } if (slot.first_disconnect_time > 0) { unsigned long first_disc_age_sec = (millis() - slot.first_disconnect_time) / 1000; replyAppendf(buf, bufsize, &pos, ", first_disc:%lus", first_disc_age_sec); } } + // cf: attempts that never reached onConnect, plus starts that failed outright. A + // slot retrying into a heap that cannot satisfy the TLS allocation moves only this; + // the publish counters cannot move at all while the slot is down. + const uint32_t cf = slot.connect_failures + slot.start_failures; + if (cf > 0) { + replyAppendf(buf, bufsize, &pos, ", cf:%lu", (unsigned long)cf); + } // Connected with no errors: nothing more to say about the connection. if (slot.connected && slot.last_error_time == 0) { @@ -1728,6 +1729,7 @@ bool MQTTBridge::ensureSlotClient(int index) { MQTT_DEBUG_PRINTLN("MQTT%d ignoring late CONNECTED (state=%s, gen=%lu, enabled=%d)", index + 1, clientStateName(st), (unsigned long)_slots[index].generation, (int)_slots[index].enabled); + _slot_attempt_pending[index] = false; // still the end of that attempt return; } MQTT_DEBUG_PRINTLN("MQTT%d connected", index + 1); @@ -1766,8 +1768,9 @@ bool MQTTBridge::ensureSlotClient(int index) { slot.client->onDisconnect([this, index](bool sessionPresent) { MQTT_DEBUG_PRINTLN("MQTT%d disconnected", index + 1); // Ended an attempt that never reached onConnect. A deliberate stop clears the flag - // first, so cancelling an attempt is not counted as a failure. - if (_slot_attempt_pending[index]) { + // first; a bounced live session is still marked connected until this handler runs, + // so its late DISCONNECTED after a timed-out softDisconnect() is not counted either. + if (_slot_attempt_pending[index] && !_slots[index].connected) { _slot_attempt_pending[index] = false; _slots[index].connect_failures++; } @@ -2524,6 +2527,7 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { slot.applied_token_expires_at = slot.token_expires_at; } else { _slot_attempt_pending[index] = false; + slot.start_failures++; } return r; } @@ -3199,7 +3203,7 @@ MQTTConnHealth MQTTBridge::collectConnHealth() const { const MQTTSlot& slot = _slots[i]; if (!slot.enabled) continue; // "none" slots are not an outage, they are unconfigured health.slots_total++; - connect_failures += slot.connect_failures; + connect_failures += slot.connect_failures + slot.start_failures; if (slot.circuit_breaker_tripped) breaker_slots++; if (slot.connected) { health.slots_up++; diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 87b75b050f..87e585b107 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -187,6 +187,10 @@ class MQTTBridge : public BridgeBase { // disconnect_count (which also counts dropped live sessions) and from the // publish error counter, which cannot move at all while a slot is down. uint32_t connect_failures; + // connect()/reconnect() calls that returned an error, so no attempt started and + // no event will arrive. Written only by the bridge task; reported summed with + // connect_failures. + uint32_t start_failures; unsigned long first_disconnect_time; // millis() of first disconnect after boot // Current-outage timer (used by AlertReporter to fire faults after a sustained @@ -288,7 +292,7 @@ class MQTTBridge : public BridgeBase { // Set by the bridge task before connect()/reconnect(), so a failure the event task // delivers before that call returns is still seen; cleared by the attempt's outcome - // or a deliberate stop. A disconnect that finds it set is a connect failure. + // (including an ignored late CONNECTED) or a deliberate stop. volatile bool _slot_attempt_pending[RUNTIME_MQTT_SLOTS]; // CLI-requested forced NTP sync, marshalled onto the MQTT task (Core 0). @@ -381,8 +385,9 @@ class MQTTBridge : public BridgeBase { // _device_id[65], _board_model[64], _firmware_version[64] and a 63-byte client version): // measured 762/768 before the connection-health fields existed. Overflow is silent — // serializeComplete() returns 0 and the bridge publishes only len>0, so status stops - // entirely. Raised to 1024, which costs no extra RAM here because status serializes into - // the shared scratch buffer. See the worst-case test in test_mqtt_payload_builder. + // entirely. Raised to 1024: no extra RAM while status serializes into the shared scratch + // buffer, but +256 B of bridge-task stack in the fallback used if the PSRAM allocation + // failed. See the worst-case test in test_mqtt_payload_builder. static const size_t STATUS_JSON_BUFFER_SIZE = 1024; static_assert(STATUS_JSON_BUFFER_SIZE <= PUBLISH_JSON_BUFFER_SIZE, "status payloads serialize into the shared publish buffer"); 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 e8ae51c871..eedf69795c 100644 --- a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp +++ b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp @@ -222,11 +222,11 @@ TEST(MQTTPayloadBuilder, WorstCaseStatusWithConnectionHealthFitsProductionBuffer char buffer[kStatusJsonBufferSize]; MQTTConnHealth health; health.slots_up = 0; - health.slots_total = 5; // PSRAM slot cap, the widest configuration + health.slots_total = 6; // RUNTIME_MQTT_SLOTS on PSRAM, the widest health.worst_outage_secs = 4294967; // millis() rollover / 1000, the widest value health.heap_largest = 8388608; health.connect_failures = 2147483647; - health.slots_breaker = 5; + health.slots_breaker = 6; int len = MQTTPayloadBuilder::buildStatusMessage( scratch, origin.c_str(), origin_id.c_str(), model.c_str(), firmware.c_str(), From f37445f8a7d14f7d75e1ec848398f591d793b6f6 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 26 Sep 2026 12:48:44 -0700 Subject: [PATCH 6/7] fix(mqtt): count only failed starts, not rejected reconnects, in cf reconnect() on a started client returns ESP_FAIL while an attempt is already in progress. On a Heltec V3, a live reconfigure hit exactly that and the attempt then connected, yet cf read 1. Count start_failures only when connect() on a stopped client fails, and restore the attempt flag to its prior value on any failed call so the in-progress attempt is still judged by its own outcome. --- src/helpers/bridges/MQTTBridge.cpp | 11 ++++++++--- src/helpers/bridges/MQTTBridge.h | 6 +++--- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 2bee5bb250..8e5eec70a7 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -2512,8 +2512,10 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { } esp_err_t r; + const bool was_pending = _slot_attempt_pending[index]; + const bool starting = !slot.client->isStarted(); _slot_attempt_pending[index] = true; - if (!slot.client->isStarted()) { + if (starting) { MQTT_DEBUG_PRINTLN("MQTT%d start (client was stopped)", index + 1); r = slot.client->connect(); } else { @@ -2526,8 +2528,11 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { // is the point at which a freshly minted token becomes the one in use. slot.applied_token_expires_at = slot.token_expires_at; } else { - _slot_attempt_pending[index] = false; - slot.start_failures++; + _slot_attempt_pending[index] = was_pending; + // A failed start (buffer/task allocation, config) means no attempt could begin. + // reconnect() on a started client returns ESP_FAIL while one is already in + // progress, which is not a failure (seen on hardware after a reconfigure). + if (starting) slot.start_failures++; } return r; } diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 87e585b107..767a49e10f 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -187,9 +187,9 @@ class MQTTBridge : public BridgeBase { // disconnect_count (which also counts dropped live sessions) and from the // publish error counter, which cannot move at all while a slot is down. uint32_t connect_failures; - // connect()/reconnect() calls that returned an error, so no attempt started and - // no event will arrive. Written only by the bridge task; reported summed with - // connect_failures. + // connect() calls on a stopped client that returned an error, so no attempt + // started and no event will arrive. Written only by the bridge task; reported + // summed with connect_failures. uint32_t start_failures; unsigned long first_disconnect_time; // millis() of first disconnect after boot From 7d0a602b20cd61419005566bce5b5da7cb644160 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 26 Sep 2026 13:54:43 -0700 Subject: [PATCH 7/7] fix(mqtt): exclude unready slots from slot health and keep cf monotonic A slot waiting on a token/IATA/credential stays enabled but never attempts a connection, so it held mqtt_slots_up below mqtt_slots_total (and down= at 1) permanently -- a standing false alert on a node that is merely unconfigured. slots_total and down= now count only ready slots, matching the CLI's "wait". mqtt_connect_failures summed only enabled slots, so disabling a failing slot made it drop. It now sums every slot, monotonic within a boot. cf: moves after the error tail in mqttN.diag so a clamped 160-byte reply loses the count rather than the refusal/TLS detail. A failed start now clears the attempt flag instead of restoring a possibly stale one. Outage comments corrected: the clock starts at a slot's first DISCONNECTED, not first connect. --- src/helpers/MQTTPayloadBuilder.cpp | 2 +- src/helpers/MQTTPayloadBuilder.h | 10 ++++----- src/helpers/bridges/MQTTBridge.cpp | 36 +++++++++++++++++------------- 3 files changed, 26 insertions(+), 22 deletions(-) diff --git a/src/helpers/MQTTPayloadBuilder.cpp b/src/helpers/MQTTPayloadBuilder.cpp index 8cbb03f4ce..56e2a259db 100644 --- a/src/helpers/MQTTPayloadBuilder.cpp +++ b/src/helpers/MQTTPayloadBuilder.cpp @@ -88,7 +88,7 @@ int MQTTPayloadBuilder::buildStatusMessage( 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. + // health: a slot has no outage start time until its first attempt ends. // `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; diff --git a/src/helpers/MQTTPayloadBuilder.h b/src/helpers/MQTTPayloadBuilder.h index 7f0cd85b0f..fa46e7415b 100644 --- a/src/helpers/MQTTPayloadBuilder.h +++ b/src/helpers/MQTTPayloadBuilder.h @@ -21,12 +21,12 @@ // per-attempt reading. 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 slots_total = -1; // enabled, fully configured slots, connected or not + int worst_outage_secs = -1; // longest timed outage; -1 = none timed (see below) int heap_largest = -1; // largest free internal block, bytes - // Attempts that ended without reaching onConnect, summed across slots. A slot - // stuck retrying moves this while its publish counters stay frozen, which is - // what separates "actively failing" from "idle and quiet". + // Attempts that ended without reaching onConnect, summed across all slots since + // boot. A slot stuck retrying moves this while its publish counters stay frozen, + // which is what separates "actively failing" from "idle and quiet". int connect_failures = -1; // Slots parked by the circuit breaker. Materially worse than a retrying slot: // the breaker only probes every 30 minutes, so recovery is far slower. diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 8e5eec70a7..8c24fbf2bc 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -373,13 +373,13 @@ 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 + // down=: ready 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 (b->_slots[i].enabled && !b->_slots[i].connected && b->isSlotReady(i)) down++; } if (down > 0) { replyAppendf(buf, bufsize, &pos, " down=%d", down); @@ -589,13 +589,6 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) replyAppendf(buf, bufsize, &pos, ", first_disc:%lus", first_disc_age_sec); } } - // cf: attempts that never reached onConnect, plus starts that failed outright. A - // slot retrying into a heap that cannot satisfy the TLS allocation moves only this; - // the publish counters cannot move at all while the slot is down. - const uint32_t cf = slot.connect_failures + slot.start_failures; - if (cf > 0) { - replyAppendf(buf, bufsize, &pos, ", cf:%lu", (unsigned long)cf); - } // Connected with no errors: nothing more to say about the connection. if (slot.connected && slot.last_error_time == 0) { @@ -642,6 +635,14 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) } else if (!slot.connected) { replyAppendf(buf, bufsize, &pos, ", no error info"); } + // cf: attempts that never reached onConnect, plus starts that failed outright. A + // slot retrying into a heap that cannot satisfy the TLS allocation moves only this; + // the publish counters cannot move at all while the slot is down. + // After the error tail so a clamped reply loses this count, not the error detail. + const uint32_t cf = slot.connect_failures + slot.start_failures; + if (cf > 0) { + replyAppendf(buf, bufsize, &pos, ", cf:%lu", (unsigned long)cf); + } // Appended last so it never displaces connection diagnostics. replyAppendf // clamps rather than overflows, but a clipped type list is worse than no @@ -2528,7 +2529,8 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { // is the point at which a freshly minted token becomes the one in use. slot.applied_token_expires_at = slot.token_expires_at; } else { - _slot_attempt_pending[index] = was_pending; + // A failed start leaves nothing in flight; a rejected reconnect leaves the prior one. + _slot_attempt_pending[index] = starting ? false : was_pending; // A failed start (buffer/task allocation, config) means no attempt could begin. // reconnect() on a started client returns ESP_FAIL while one is already in // progress, which is not a failure (seen on hardware after a reconfigure). @@ -3206,19 +3208,21 @@ MQTTConnHealth MQTTBridge::collectConnHealth() const { 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++; + // Summed before the skip so disabling a slot never makes the total go backwards. connect_failures += slot.connect_failures + slot.start_failures; + // Unconfigured slots and slots waiting on a token/IATA/credential are not outages. + if (!slot.enabled || !isSlotReady(i)) continue; + health.slots_total++; if (slot.circuit_breaker_tripped) breaker_slots++; 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. + // The outage clock starts at a slot's first DISCONNECTED, so a slot that has not yet + // finished its first attempt has no 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 + const unsigned long outage_ms = now - slot.current_outage_started_ms; // wrap-safe under 49.7 days if (!any_outage_timed || outage_ms > worst_outage_ms) { worst_outage_ms = outage_ms; any_outage_timed = true;