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:<n>` beside the existing `dc:<n>`.
- 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

@@ -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);

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.

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<int>(worst_outage_ms / 1000UL);
   }
+  // Clamp: the payload field is int, and these are lifetime counters.
+  health.connect_failures = connect_failures > static_cast<uint32_t>(INT32_MAX)
+                                ? INT32_MAX
+                                : static_cast<int>(connect_failures);
+  health.slots_breaker = breaker_slots;
   health.heap_largest = static_cast<int>(heap_caps_get_largest_free_block(MALLOC_CAP_INTERNAL));
   return health;
 }

src/helpers/bridges/MQTTBridge.h

@@ -183,6 +183,10 @@ private:
     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

test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp

@@ -118,6 +118,46 @@ TEST(MQTTPayloadBuilder, StatusReportsConnectionHealthWhenSupplied) {
   EXPECT_EQ(16116, stats["heap_largest"].as<int>());
 }
 
+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<JsonObject>();
+  EXPECT_EQ(57, stats["mqtt_connect_failures"].as<int>());
+  EXPECT_EQ(1, stats["mqtt_slots_breaker"].as<int>());
+
+  // 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<JsonObject>();
+  EXPECT_FALSE(healthy_stats["mqtt_connect_failures"].is<JsonVariant>());
+  EXPECT_FALSE(healthy_stats["mqtt_slots_breaker"].is<JsonVariant>());
+}
+
 // 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(),