mirror of
https://github.com/agessaman/MeshCore.git
synced 2026-10-24 04:02:39 +00:00
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=<n>` (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.
This commit is contained in:
1 parent
43b55de26f
commit
f7ed2f176d
7 files changed
+184
-10
No files matched your search
@@ -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(
|
||||
|
||||
@@ -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()
|
||||
);
|
||||
|
||||
/**
|
||||
|
||||
@@ -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<JsonObject>();
|
||||
@@ -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<JsonObject>();
|
||||
|
||||
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);
|
||||
|
||||
@@ -4,6 +4,26 @@
|
||||
#include <stddef.h>
|
||||
#include <stdint.h>
|
||||
|
||||
// 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(
|
||||
|
||||
@@ -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=<n>: 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<int>(worst_outage_ms / 1000UL);
|
||||
}
|
||||
health.heap_largest = static_cast<int>(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) {
|
||||
|
||||
@@ -9,6 +9,7 @@
|
||||
#include <Timezone.h>
|
||||
#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();
|
||||
|
||||
@@ -93,6 +93,85 @@ TEST(MQTTPayloadBuilder, StatusIncludesRepeatAndEveryRequestedStatistic) {
|
||||
EXPECT_EQ(47, stats["packets_received"].as<int>());
|
||||
}
|
||||
|
||||
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<JsonObject>();
|
||||
ASSERT_FALSE(stats.isNull());
|
||||
EXPECT_EQ(2, stats["mqtt_slots_up"].as<int>());
|
||||
EXPECT_EQ(3, stats["mqtt_slots_total"].as<int>());
|
||||
EXPECT_EQ(18432, stats["mqtt_outage_secs"].as<int>());
|
||||
EXPECT_EQ(16116, stats["heap_largest"].as<int>());
|
||||
}
|
||||
|
||||
// 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<JsonObject>();
|
||||
ASSERT_FALSE(stats.isNull());
|
||||
EXPECT_EQ(3, stats["mqtt_slots_up"].as<int>());
|
||||
EXPECT_FALSE(stats["mqtt_outage_secs"].is<JsonVariant>());
|
||||
}
|
||||
|
||||
// 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<size_t>(len), kStatusJsonBufferSize - 32)
|
||||
<< "less than 32 bytes of headroom left in the status buffer";
|
||||
}
|
||||
|
||||
TEST(MQTTPayloadBuilder, StatusOmissionSentinelsRemainOmitted) {
|
||||
JsonDocument scratch;
|
||||
char buffer[768];
|
||||
|
||||
Reference in new issue
Block a user