From 3347a62780a7eeef55cd5571390170c1a603c275 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Wed, 14 Oct 2026 10:34:40 -0700 Subject: [PATCH] Recover stalled MQTT transports and broker acknowledgments --- .github/workflows/run-unit-tests.yml | 1 + docs/WiFi.md | 14 + docs/cli_command_availability.md | 9 + docs/test-results/mqtt_tls_v4_regression.md | 66 +++ src/helpers/MQTTTransportHealth.h | 79 +++ src/helpers/bridges/MQTTBridge.cpp | 255 +++++++++- src/helpers/bridges/MQTTBridge.h | 15 + test/README.md | 1 + test/fixtures/mqtt_ap_ownership.cpp | 11 +- test/test_mqtt_connection_admission.py | 3 + test/test_mqtt_disconnect.py | 6 + test/test_mqtt_handshake_scheduler.py | 506 +++++++++++++++++++- test/test_mqtt_network_timeout.py | 6 + test/test_mqtt_publish_ack.py | 52 ++ test/test_mqtt_transport_health.py | 134 ++++++ test/test_mqtt_transport_results.py | 4 + test/test_mqtt_wifi_network_power.py | 62 ++- 17 files changed, 1201 insertions(+), 23 deletions(-) create mode 100644 src/helpers/MQTTTransportHealth.h create mode 100644 test/test_mqtt_transport_health.py diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index 501f85561..f8d409638 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -246,6 +246,7 @@ jobs: python3 -B test/test_mqtt_disconnect.py -v python3 -B test/test_mqtt_connection_admission.py -v python3 -B test/test_mqtt_handshake_scheduler.py -v + python3 -B test/test_mqtt_transport_health.py -v python3 -B test/test_mqtt_diagnostics.py -v python3 -B test/test_mqtt_publish_ack.py -v python3 -B test/test_wifi_network_power_policy.py -v diff --git a/docs/WiFi.md b/docs/WiFi.md index 694b375db..42b5be403 100644 --- a/docs/WiFi.md +++ b/docs/WiFi.md @@ -734,6 +734,20 @@ get mqtt1.diag get mqtt2.diag ``` +For a connection that works initially but later stops reporting, inspect each +configured slot's `pub_ack` count in `get mqttN.diag`. Accepted publishes are +not proof of delivery: a QoS 1 publish is confirmed only when the broker +acknowledges it. A quiet radio channel alone is not a failed MQTT connection. + +The shared MQTT bridge recovers failed transports on its owning task, including +errors without a subsequent disconnect callback and connection attempts that +never finish. It also watches accepted QoS 1 publishes for missing broker +acknowledgments. Recovery retains the existing broker configuration, memory +admission and reconnect backoff; it does not reboot the node or cycle WiFi. +See the [MQTT recovery regression](test-results/mqtt_tls_v4_regression.md#long-running-mqtt-recovery) +for the deadlines, test coverage and hardware limitations. Older preview images +do not gain these changes until their firmware is updated. + Common causes are: - no saved SSID or a changed password; diff --git a/docs/cli_command_availability.md b/docs/cli_command_availability.md index 882d5d7e5..e2f6d9e0b 100644 --- a/docs/cli_command_availability.md +++ b/docs/cli_command_availability.md @@ -518,6 +518,15 @@ clear, set, or toggle selected bits. Prints the runtime diagnostic summary for MQTT slot N, including its connection state and failure history. +`pub_ack` counts actual broker acknowledgments, not merely accepted writes. +Recent firmware also shows `rec:/` for transport recovery and +`ack_wait:s` while an accepted QoS 1 publication awaits acknowledgment. +Reasons include `error`, `state`, `handshake`, `write`, and `puback`. `recover` +means cleanup is pending, not that the broker is healthy. Optional fields are +omitted if they cannot fit completely in the reply. `rec` counts completed +client cleanup; a higher count alone does not prove the subsequent connection +or delivery succeeded. + ### `get mqtt.stats` diff --git a/docs/test-results/mqtt_tls_v4_regression.md b/docs/test-results/mqtt_tls_v4_regression.md index d179d252f..68ff349ad 100644 --- a/docs/test-results/mqtt_tls_v4_regression.md +++ b/docs/test-results/mqtt_tls_v4_regression.md @@ -199,3 +199,69 @@ Comparison removes console echo/prompt framing and treats the old blank `get wifi.ssid` response as the new `(not configured)` response. MQTT and power saving were both returned to their original off state. No release assets were uploaded as part of this test. + +## Long-running MQTT recovery + +The [14 October 2026 follow-up report](https://github.com/mikecarper/meshfirmware/issues/16#issuecomment-6169059046) +says the preview initially connects, but three reporting sites show it offline +after roughly three hours while WiFi and the console still show connected. +The earlier 180-second trial does not qualify this longer-running workload. + +Code inspection found recovery gaps, not a verified cause of that hardware +failure. The adapter clears its connection flag on an MQTT error, but the +bridge previously left its separate flag set until a disconnect callback. +Its five-second cache refresh only reread that same flag. A lost terminal +callback could also leave a connection attempt pending indefinitely, blocking +the other brokers' handshake and renewal requests. Accepted status enqueues +were treated as successful publication without using the existing PUBACK +counter to detect stalled delivery. + +Recovery now runs on the bridge's owning task. An errored or timed-out SDK +client must be proven stopped before its pending-handshake reservation is +released. The client object and credentials are retained, and the existing +admission checks and reconnect backoff still apply. No WiFi reset, node reboot, +extra periodic heartbeat, or broker credential change is required. An idle +connection with no outstanding QoS 1 publication is not timed out merely for +being quiet. + +Both the pending-handshake deadline and the outstanding-PUBACK deadline are +120 seconds. This leaves a generous margin over the SDK's 10-second network +timeout and the configured 15-second QoS 1 retransmission interval. A failed +SDK stop retains its recovery reservation instead of permitting another +overlapping TLS handshake. + +CLI and WebConfig slot status show `recover` during reserved cleanup. +`get mqttN.diag` retains `pub_ack` and adds a bounded recovery count/reason +(`rec`) and pending acknowledgment age (`ack_wait`) when they fit. A delayed +connected callback cannot cancel cleanup after an SDK stop failed. + +The delivery watch starts with the first accepted QoS 1 enqueue and observes +real broker acknowledgments, including the on-connect status publication. +Further enqueues cannot postpone that oldest outstanding deadline. Timeouts +are serviced on the next owning-task pass; they are not a guarantee of +wall-clock recovery if the worker is blocked inside an SDK call. This change +does not establish or repair an indefinite SDK mutex deadlock. + +All 251 native MQTT test cases and 79 MQTT Python host tests passed. The new +tests execute the production callbacks, publishing, maintenance, recovery and +admission methods with controllable SDK calls and clocks. They include three +brokers running beyond three simulated hours, missing PUBACKs and callbacks, +silent synchronous write failures, failed SDK stops, late connected events, +WiFi loss, suspension/foreground wake, backoff, memory admission and timer +rollover. Removing the new error-callback recovery signal makes its regression +test fail, establishing that it detects the previous behavior. AP ownership +fixtures also passed with address/undefined-behavior sanitizers. The six pure +transport-health tests are included in the 79-test total and in CI. + +The exact Full `heltec_v4_repeater_observer_mqtt` image compiled and linked +against Arduino 2.0.17 / ESP-IDF 4.4.7. This local test image used baseline +`dce323e37` with the MQTT repair applied; it predates the subsequently pulled +beacon changes. Its application size was 2,237,677 +bytes, within the 6,553,600-byte application slot. The linked runtime RAM gate +reported 204,752 internal bytes available, 198,432 required and a 171,984-byte +largest region. These are build-time capacity/admission checks, not free heap +measured with three live brokers or qualification of maximum simultaneous load. + +No new physical three-hour V4 soak or reporter-side retest has been performed +as part of this repair. These tests establish the repaired recovery behavior, +not the cause of the reporter's hardware failure or website-side availability. diff --git a/src/helpers/MQTTTransportHealth.h b/src/helpers/MQTTTransportHealth.h new file mode 100644 index 000000000..3bc4b3384 --- /dev/null +++ b/src/helpers/MQTTTransportHealth.h @@ -0,0 +1,79 @@ +#pragma once + +#include + +// Owner-task transport recovery, separate from the reconnect/backoff policy. +// A CONNECTED callback or accepted/enqueued publish is not proof of delivery. +namespace MQTTTransportHealth { + +// Twelve times the normal 10-second socket budget gives DNS/TLS/WebSocket/ +// CONNACK setup ample room, while bounding a lost final callback. PUBACK uses +// the same generous budget: longer than the 15-second retry and SDK outbox +// expiry, so an ordinary retransmit cannot trigger a premature reconnect. +static const uint32_t kAttemptTimeoutMs = 120000UL; +static const uint32_t kAckTimeoutMs = 120000UL; + +enum class RecoveryReason : uint8_t { + None, TransportError, AdapterDisconnected, HandshakeTimeout, + PublishFailure, AckTimeout +}; + +inline const char* reasonLabel(RecoveryReason reason) { + switch (reason) { + case RecoveryReason::TransportError: return "error"; + case RecoveryReason::AdapterDisconnected: return "state"; + case RecoveryReason::HandshakeTimeout: return "handshake"; + case RecoveryReason::PublishFailure: return "write"; + case RecoveryReason::AckTimeout: return "puback"; + default: return "none"; + } +} + +struct State { + bool attempt_active = false; + uint32_t attempt_started_ms = 0; + bool ack_pending = false; + uint32_t ack_started_ms = 0; + uint32_t ack_snapshot = 0; + RecoveryReason recovery_reason = RecoveryReason::None; + RecoveryReason last_recovery_reason = RecoveryReason::None; + uint32_t recoveries = 0; + + void beginAttempt(uint32_t now) { + attempt_active = true; + attempt_started_ms = now; + } + void completeAttempt() { attempt_active = false; } + void observeAck(uint32_t count) { + if (ack_pending && count != ack_snapshot) ack_pending = false; + } + void acceptedQos1(uint32_t now, uint32_t ack_before) { + observeAck(ack_before); + if (!ack_pending) { + ack_pending = true; + ack_started_ms = now; + ack_snapshot = ack_before; + } + } + bool attemptExpired(uint32_t now) const { + return attempt_active && now - attempt_started_ms >= kAttemptTimeoutMs; + } + bool ackExpired(uint32_t now) const { + return ack_pending && now - ack_started_ms >= kAckTimeoutMs; + } + void requestRecovery(RecoveryReason reason) { + if (recovery_reason == RecoveryReason::None) recovery_reason = reason; + } + void resetSession() { + attempt_active = false; + ack_pending = false; + } + void recovered() { + if (recoveries != UINT32_MAX) ++recoveries; + last_recovery_reason = recovery_reason; + recovery_reason = RecoveryReason::None; + resetSession(); + } +}; + +} // namespace MQTTTransportHealth diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 21754a061..139923ca1 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -426,6 +426,10 @@ void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPref } else if (!b->isSlotReady(i)) { name = slot.preset ? slot.preset->name : "custom"; state = "wait"; + } else if (b->_slot_health_recovering[i].load(std::memory_order_acquire) || + b->_slot_transport_fault_pending[i].load(std::memory_order_acquire)) { + name = slot.preset ? slot.preset->name : "custom"; + state = "recover"; } else if (slot.connected) { name = slot.preset ? slot.preset->name : "custom"; state = "ok"; @@ -551,6 +555,9 @@ bool MQTTBridge::getSlotStatusSnapshot(int slot_index, SlotStatusSnapshot* out) out->name = slot.preset ? slot.preset->name : "custom"; if (!b->isSlotReady(slot_index)) { out->state = "wait"; + } else if (b->_slot_health_recovering[slot_index].load(std::memory_order_acquire) || + b->_slot_transport_fault_pending[slot_index].load(std::memory_order_acquire)) { + out->state = "recover"; } else if (slot.connected) { out->state = "ok"; } else if (slot.circuit_breaker_tripped) { @@ -657,6 +664,9 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) } else if (!slot.client) { // Ready to connect but the client object could not be allocated. state = "no client"; + } else if (b->_slot_health_recovering[slot_index].load(std::memory_order_acquire) || + b->_slot_transport_fault_pending[slot_index].load(std::memory_order_acquire)) { + state = "recover"; } else if (slot.connected) { state = "ok"; } else if (slot.circuit_breaker_tripped) { @@ -689,7 +699,7 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) } // Connected with no errors: nothing more to say about the connection. - if (slot.connected && slot.last_error_time == 0) { + if (slot.connected && strcmp(state, "ok") == 0 && slot.last_error_time == 0) { replyAppendf(buf, bufsize, &pos, ", no errors"); } else if (slot.last_error_time > 0) { // TLS error with human-friendly description @@ -743,6 +753,28 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) } } + // Recovery/ACK age are owner-task snapshots. Append complete fields only; + // error details and the existing broker ACK count retain reply priority. + const uint32_t recoveries = b->_slot_health_recoveries[slot_index].load(std::memory_order_relaxed); + const auto recovery_reason = static_cast( + b->_slot_health_reason[slot_index].load(std::memory_order_relaxed)); + if (recoveries || recovery_reason != MQTTTransportHealth::RecoveryReason::None) { + char recovery_text[48]; + const int len = snprintf(recovery_text, sizeof(recovery_text), ", rec:%lu/%s", + static_cast(recoveries), MQTTTransportHealth::reasonLabel(recovery_reason)); + if (len > 0 && len < (int)sizeof(recovery_text) && len <= (int)bufsize - 1 - pos) { + replyAppendf(buf, bufsize, &pos, "%s", recovery_text); + } + } + const uint32_t ack_wait = b->_slot_health_ack_wait_secs[slot_index].load(std::memory_order_relaxed); + if (ack_wait != UINT32_MAX) { + char wait_text[32]; + const int len = snprintf(wait_text, sizeof(wait_text), ", ack_wait:%lus", static_cast(ack_wait)); + if (len > 0 && len < (int)sizeof(wait_text) && len <= (int)bufsize - 1 - pos) { + replyAppendf(buf, bufsize, &pos, "%s", wait_text); + } + } + // Appended last so it never displaces connection diagnostics. replyAppendf // clamps rather than overflows, but a clipped type list is worse than no // list: "...,13,14," parses as a real, different allowlist, and this is the @@ -875,11 +907,11 @@ MQTTBridge::MQTTBridge(const MQTTNodeInfo& node_info, MQTTPrefs *obs, // Initialize all slots to empty/disabled state for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - memset(&_slots[i], 0, sizeof(MQTTSlot)); + _slots[i] = MQTTSlot{}; _slots[i].enabled = false; _slots[i].client = nullptr; _slots[i].preset = nullptr; - // auth_token == nullptr after memset above - allocated on first token creation + // auth_token == nullptr after value initialization - allocated on first token creation _slots[i].connected = false; _slots[i].initial_connect_done = false; _slots[i].token_expires_at = 0; @@ -893,6 +925,13 @@ MQTTBridge::MQTTBridge(const MQTTNodeInfo& node_info, MQTTPrefs *obs, _slot_reconfigure_pending[i] = false; _slot_force_jwt_mint[i] = false; _slot_attempt_pending[i] = false; + _slot_transport_fault_pending[i] = false; + _slot_health_reset_pending[i] = false; + _slots[i].transport_health = MQTTTransportHealth::State(); + _slot_health_recoveries[i] = 0; + _slot_health_reason[i] = static_cast(MQTTTransportHealth::RecoveryReason::None); + _slot_health_recovering[i] = false; + _slot_health_ack_wait_secs[i] = UINT32_MAX; _status_publish_pending[i] = false; } @@ -1024,6 +1063,11 @@ void MQTTBridge::begin() { _slots[i].broker_uri[0] = '\0'; _slots[i].port = 1883; _slot_reconfigure_pending[i] = false; + _slot_transport_fault_pending[i] = false; + _slot_health_reset_pending[i] = false; + _slots[i].transport_health.resetSession(); + _slots[i].transport_health.recovery_reason = MQTTTransportHealth::RecoveryReason::None; + updateSlotTransportDiag(i, static_cast(millis())); } _cached_has_connected_slots = false; _slots_setup_done = false; @@ -1617,13 +1661,33 @@ bool MQTTBridge::serviceNetworkPower(uint32_t now) { cancelNtpRefresh(); _ntp_sync_pending = false; for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { + // A foreground/probe request can resume the network policy after a + // failed suspension. Keep a transport reservation until stop is proven. + if (_slots[i].client && _slots[i].client->isStarted()) { + _slot_transport_fault_pending[i].store(true, std::memory_order_release); + } _slot_attempt_pending[i] = false; if (_slots[i].client && _slots[i].client->isStarted()) { _slots[i].client->disconnect(); // The wrapper only clears isStarted after SDK stop succeeds. Never // hand station ownership off while any transport still runs. - if (_slots[i].client->isStarted()) return false; + if (_slots[i].client->isStarted()) { + _slots[i].transport_health.requestRecovery(MQTTTransportHealth::RecoveryReason::AdapterDisconnected); + _slots[i].connected = false; + _slots[i].connected_at_ms = 0; + if (_slots[i].current_outage_started_ms == 0) { + _slots[i].current_outage_started_ms = now ? now : 1; + } + updateSlotTransportDiag(i, static_cast(millis())); + updateCachedConnectionStatus(); + return false; + } } + _slot_transport_fault_pending[i].store(false, std::memory_order_release); + _slot_health_reset_pending[i].store(false, std::memory_order_release); + _slots[i].transport_health.resetSession(); + _slots[i].transport_health.recovery_reason = MQTTTransportHealth::RecoveryReason::None; + updateSlotTransportDiag(i, static_cast(millis())); _slots[i].connected = false; _slots[i].connected_at_ms = 0; _status_publish_pending[i] = false; @@ -2040,8 +2104,8 @@ void MQTTBridge::mqttTaskLoop() { } } - // Update cached connection status periodically (every 5 seconds) - // This ensures cache stays accurate even if callbacks miss updates + // Refresh the aggregate cache periodically; owner maintenance above + // reconciles adapter state and bounds missed terminal callbacks. static unsigned long last_slot_status_update = 0; if (now - last_slot_status_update > 5000) { updateCachedConnectionStatus(); @@ -2092,6 +2156,7 @@ bool MQTTBridge::ensureSlotClient(int index) { MQTT_DEBUG_PRINTLN("MQTT%d connected", index + 1); _slot_attempt_pending[index] = false; _slots[index].connected = true; + _slot_health_reset_pending[index].store(true, std::memory_order_release); _slot_force_jwt_mint[index] = false; // NOTE: reconnect_backoff / max_backoff_failures are NOT reset here. // A CONNACK alone doesn't prove the link is healthy -- a broker that @@ -2142,6 +2207,10 @@ bool MQTTBridge::ensureSlotClient(int index) { updateCachedConnectionStatus(); }); slot.client->onError([this, index](esp_mqtt_error_codes error) { + // The adapter is already disconnected at MQTT_EVENT_ERROR. Do not stop + // its SDK task from inside that task's callback, or release the global + // handshake reservation before the owner proves the SDK has stopped. + _slot_transport_fault_pending[index].store(true, std::memory_order_release); _slots[index].last_tls_err = error.esp_tls_last_esp_err; _slots[index].last_tls_stack_err = error.esp_tls_stack_err; _slots[index].last_sock_errno = error.esp_transport_sock_errno; @@ -2255,7 +2324,9 @@ bool MQTTBridge::hasPendingSlotConnection() const { // Callbacks run on esp-mqtt tasks. Time spacing alone cannot establish that // an earlier handshake has finished, particularly on a slow WiFi link. for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { - if (_slot_attempt_pending[i].load(std::memory_order_acquire)) return true; + if (_slot_attempt_pending[i].load(std::memory_order_acquire) || + _slot_transport_fault_pending[i].load(std::memory_order_acquire) || + _slots[i].transport_health.recovery_reason != MQTTTransportHealth::RecoveryReason::None) return true; } return false; } @@ -2326,6 +2397,9 @@ bool MQTTBridge::setupSlot(int index) { slot.client->disconnect(); } if (slot.client->isStarted()) { + _slot_transport_fault_pending[index].store(true, std::memory_order_release); + slot.transport_health.requestRecovery(MQTTTransportHealth::RecoveryReason::AdapterDisconnected); + updateSlotTransportDiag(index, static_cast(millis())); slot.last_reconnect_attempt = millis(); return false; } @@ -2558,7 +2632,14 @@ void MQTTBridge::teardownSlot(int index, bool force) { delay(50); #endif } - if (slot.client && !force && slot.client->isStarted()) return; + if (slot.client && !force && slot.client->isStarted()) { + // Even a deliberate reconfigure must retain admission ownership if SDK + // stop failed. The worker will retry cleanup; no second handshake may run. + _slot_transport_fault_pending[index].store(true, std::memory_order_release); + slot.transport_health.requestRecovery(MQTTTransportHealth::RecoveryReason::AdapterDisconnected); + updateSlotTransportDiag(index, static_cast(millis())); + return; + } // Invalidate the token but keep the buffer: the client survives teardown and still // holds this pointer in its config (see MQTTSlot::auth_token). @@ -2575,6 +2656,11 @@ void MQTTBridge::teardownSlot(int index, bool force) { slot.last_log_time = 0; slot.last_deferred_log_ms = 0; _slot_force_jwt_mint[index] = false; + _slot_transport_fault_pending[index] = false; + _slot_health_reset_pending[index] = false; + slot.transport_health.resetSession(); + slot.transport_health.recovery_reason = MQTTTransportHealth::RecoveryReason::None; + updateSlotTransportDiag(index, static_cast(millis())); } esp_err_t MQTTBridge::reconnectSlotClient(int index) { @@ -2586,6 +2672,7 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { const bool was_pending = _slot_attempt_pending[index]; const bool starting = !slot.client->isStarted(); + slot.transport_health.beginAttempt(static_cast(millis())); _slot_attempt_pending[index] = true; // callbacks may run before the call returns esp_err_t result; if (starting) { @@ -2597,6 +2684,7 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { if (result != ESP_OK) { if (MQTTConnectionHealth::clearPendingAfterRejectedRequest(starting, was_pending)) { _slot_attempt_pending[index] = false; + slot.transport_health.completeAttempt(); } if (starting) { slot.start_failures = MQTTConnectionHealth::incrementFailures(slot.start_failures); @@ -2605,8 +2693,112 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { return result; } +void MQTTBridge::updateSlotTransportDiag(int index, uint32_t now) { + const MQTTTransportHealth::State& health = _slots[index].transport_health; + _slot_health_recoveries[index].store(health.recoveries, std::memory_order_relaxed); + const auto reason = health.recovery_reason != MQTTTransportHealth::RecoveryReason::None + ? health.recovery_reason : health.last_recovery_reason; + _slot_health_reason[index].store(static_cast(reason), std::memory_order_relaxed); + _slot_health_recovering[index].store( + health.recovery_reason != MQTTTransportHealth::RecoveryReason::None, + std::memory_order_release); + _slot_health_ack_wait_secs[index].store( + health.ack_pending ? (now - health.ack_started_ms) / 1000UL : UINT32_MAX, + std::memory_order_relaxed); +} + +bool MQTTBridge::recoverSlotTransport(int index, uint32_t now) { + MQTTSlot& slot = _slots[index]; + auto& health = slot.transport_health; + if (!slot.client || health.recovery_reason == MQTTTransportHealth::RecoveryReason::None) return false; + + MQTT_DEBUG_PRINTLN("MQTT%d transport recovery: %s", index + 1, + MQTTTransportHealth::reasonLabel(health.recovery_reason)); + const bool failed_attempt = MQTTConnectionHealth::disconnectFailedAttempt( + _slot_attempt_pending[index].load(std::memory_order_acquire), slot.connected); + const uint32_t failures_before_stop = slot.connect_failures; + // Retain the client, SDK buffers and credentials. disconnect() closes the + // transport and stops its task, and leaves isStarted() true if SDK stop fails. + // Never vTaskDelete a live SDK task or admit a second TLS handshake on failure. + slot.client->disconnect(); + if (slot.client->isStarted()) { + updateSlotTransportDiag(index, static_cast(millis())); + return false; + } + + // SDK stop can consume its socket budget. Space the next handshake from + // completed cleanup, not from the earlier watchdog observation. + now = static_cast(millis()); + + // A real DISCONNECTED callback may already have accounted this failure. + // ERROR-only/lost-callback attempts need the same accounting, once. + if (failed_attempt && slot.connect_failures == failures_before_stop) { + slot.connect_failures = MQTTConnectionHealth::incrementFailures(slot.connect_failures); + } + _slot_attempt_pending[index].store(false, std::memory_order_release); + _slot_transport_fault_pending[index].store(false, std::memory_order_release); + _slot_health_reset_pending[index].store(false, std::memory_order_release); + _status_publish_pending[index] = false; + slot.connected = false; + slot.connected_at_ms = 0; + if (slot.current_outage_started_ms == 0) slot.current_outage_started_ms = now ? now : 1; + slot.last_reconnect_attempt = now; + _last_slot_reconnect_ms = now; + health.recovered(); + updateSlotTransportDiag(index, now); + updateCachedConnectionStatus(); + return true; +} + +void MQTTBridge::serviceSlotTransportHealth(uint32_t now) { + if (_stop_requested.load(std::memory_order_acquire)) return; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { + MQTTSlot& slot = _slots[i]; + auto& health = slot.transport_health; + if (_slot_health_reset_pending[i].exchange(false, std::memory_order_acq_rel)) { + // A late CONNACK does not release an already-required SDK stop. Its + // session clocks may reset, but only proven cleanup clears recovery. + health.resetSession(); + } + const bool pending = _slot_attempt_pending[i].load(std::memory_order_acquire); + if (!pending) health.completeAttempt(); + if (slot.client) { + health.observeAck(static_cast(slot.client->getPublishAck())); + if (_slot_transport_fault_pending[i].load(std::memory_order_acquire)) { + health.requestRecovery(MQTTTransportHealth::RecoveryReason::TransportError); + } else if (slot.connected && !slot.client->connected()) { + health.requestRecovery(MQTTTransportHealth::RecoveryReason::AdapterDisconnected); + } else if (pending && health.attemptExpired(now)) { + health.requestRecovery(MQTTTransportHealth::RecoveryReason::HandshakeTimeout); + } else if (slot.connected && health.ackExpired(now)) { + health.requestRecovery(MQTTTransportHealth::RecoveryReason::AckTimeout); + } + if (!slot.connected && !pending) health.resetSession(); + } + if (health.recovery_reason != MQTTTransportHealth::RecoveryReason::None) { + // Stop advertising a healthy link immediately, including while SDK + // cleanup is still failing. Admission is held by the recovery state, + // independently of this observable connection flag. + slot.connected = false; + slot.connected_at_ms = 0; + if (slot.current_outage_started_ms == 0) slot.current_outage_started_ms = now ? now : 1; + } + updateSlotTransportDiag(i, now); + } + updateCachedConnectionStatus(); + // Close no more than one SDK transport per pass. A failed stop keeps its + // recovery reservation and blocks all new handshakes on subsequent passes. + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { + if (_slots[i].client && _slots[i].transport_health.recovery_reason != MQTTTransportHealth::RecoveryReason::None) { + recoverSlotTransport(i, now); + break; + } + } +} + void MQTTBridge::maintainSlotConnections() { if (_stop_requested.load(std::memory_order_acquire)) return; + serviceSlotTransportHealth(static_cast(millis())); if (!_identity) return; // Check WiFi status first @@ -3003,6 +3195,14 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload if (_stop_requested.load(std::memory_order_acquire)) return false; if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; MQTTSlot& slot = _slots[index]; + // On-connect status is processed before maintenance. Consume this new- + // session latch BEFORE enqueueing it, or maintenance would erase its ACK + // watch immediately afterwards. Only this owner task mutates health state. + if (_slot_health_reset_pending[index].exchange(false, std::memory_order_acq_rel)) { + slot.transport_health.resetSession(); + } + if (slot.transport_health.recovery_reason != MQTTTransportHealth::RecoveryReason::None || + _slot_transport_fault_pending[index].load(std::memory_order_acquire)) return false; if (!slot.client || !slot.connected) { unsigned long now = millis(); if (now - slot.last_log_time > SLOT_LOG_INTERVAL) { @@ -3020,16 +3220,23 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload // write bypasses that drain ceiling entirely and does not store in the outbox. It can // block the (Core-0, prio-1) MQTT task on a stalled socket, but only up to // network_timeout_ms (configured in optimizeMqttClientConfig); mesh RX (Core 1) and the - // WiFi/TCP stack (higher-prio system tasks) are unaffected, and a failed write flips - // the slot to disconnected so subsequent packets skip it. + // WiFi/TCP stack (higher-prio system tasks) are unaffected. A failed write reserves + // owner-task transport recovery, so subsequent packets skip the failed slot. // - QoS 1 (low-rate retained status): async, so it keeps the durable outbox + retransmit. // // Return convention: QoS 0 sync publish returns msg_id == 0 on success (no PUBACK // tracking). Negative values (-1 write/failure) are the only actual failures; the queue // retry/drop path below handles them. bool async = (qos > 0); + // Take the actual broker ACK counter before the SDK call: a fast PUBACK can + // arrive before enqueue returns. A post-call snapshot would lose that proof. + const uint32_t ack_before = static_cast(slot.client->getPublishAck()); int result = slot.client->publish(topic, qos, retained, payload, (int)payload_len, async); if (result < 0) { + if (!async) { + slot.transport_health.requestRecovery(MQTTTransportHealth::RecoveryReason::PublishFailure); + updateSlotTransportDiag(index, static_cast(millis())); + } // QoS0 packet/raw publishes are best-effort and may be retried from the // bridge queue; avoid logging transient first-attempt failures here. if (qos > 0) { @@ -3042,6 +3249,12 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload } return false; } + if (qos > 0) { + slot.transport_health.acceptedQos1(static_cast(millis()), ack_before); + // Also reflect an ACK delivered during enqueue, not just on the next tick. + slot.transport_health.observeAck(static_cast(slot.client->getPublishAck())); + updateSlotTransportDiag(index, static_cast(millis())); + } return true; } @@ -3189,8 +3402,7 @@ void MQTTBridge::publishStatusToSlot(int index) { // on-connect status must not force retain=true. Custom slots default to // non-retained here too, keeping both status paths consistent. bool use_retain = slot.preset ? slot.preset->allow_retain : false; - int result = slot.client->publish(status_topic, 1, use_retain, json_buffer, len); - if (result <= 0) { + if (!publishToSlot(index, status_topic, json_buffer, static_cast(len), use_retain, 1)) { MQTT_DEBUG_PRINTLN("MQTT%d status publish failed", index + 1); } } @@ -3453,11 +3665,28 @@ bool MQTTBridge::handleWiFiConnection(unsigned long now) { s_wifi_connected_at = 0; // Disconnect all slot clients when WiFi drops for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - if (_slots[i].client && _slots[i].connected) { + // Include pending handshakes, not just sessions that reached CONNACK. + if (_slots[i].client && _slots[i].client->isStarted()) { + _slot_transport_fault_pending[i].store(true, std::memory_order_release); _slot_attempt_pending[i] = false; _slots[i].client->disconnect(); + _slots[i].connected = false; + _slots[i].connected_at_ms = 0; + if (_slots[i].client->isStarted()) { + _slots[i].transport_health.requestRecovery(MQTTTransportHealth::RecoveryReason::AdapterDisconnected); + if (_slots[i].current_outage_started_ms == 0) { + _slots[i].current_outage_started_ms = now ? now : 1; + } + } else { + _slot_transport_fault_pending[i].store(false, std::memory_order_release); + _slot_health_reset_pending[i].store(false, std::memory_order_release); + _slots[i].transport_health.resetSession(); + _slots[i].transport_health.recovery_reason = MQTTTransportHealth::RecoveryReason::None; + } + updateSlotTransportDiag(i, static_cast(millis())); } } + updateCachedConnectionStatus(); } else if (snapshot.down) { // Backoff ladder + wrap-safe timing live in MQTTConnectionPolicy (Phase 6), // exercised by host tests. Behavior is unchanged: both the link-down diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index daa672239..72435672e 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -12,6 +12,7 @@ #include "helpers/JWTHelper.h" #include "helpers/MQTTConnectionPolicy.h" #include "helpers/MQTTConnectionHealth.h" +#include "helpers/MQTTTransportHealth.h" #include "helpers/MQTTPayloadBuilder.h" #include "helpers/MQTTPacketFilter.h" #include "helpers/MQTTPresets.h" @@ -158,6 +159,10 @@ private: // DISCONNECTED (including a failed initial attempt). first_disconnect_time is intentionally separate // so the existing 'mqttN.diag' "first_disc" semantics don't change. unsigned long current_outage_started_ms; + + // Only the bridge worker mutates this state; SDK callbacks raise atomic + // latches instead. Retained across reconnects to expose recovery history. + MQTTTransportHealth::State transport_health; }; MQTTSlot _slots[RUNTIME_MQTT_SLOTS]; @@ -260,6 +265,13 @@ private: // Set before starting/reconnecting, cleared by connect/failed-attempt/stop. // Must be set before the SDK call because its event task can callback at once. std::atomic _slot_attempt_pending[RUNTIME_MQTT_SLOTS]; + std::atomic _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS]; + std::atomic _slot_health_reset_pending[RUNTIME_MQTT_SLOTS]; + // Read-only CLI snapshots: never read the owner-only State across tasks. + std::atomic _slot_health_recoveries[RUNTIME_MQTT_SLOTS]; + std::atomic _slot_health_reason[RUNTIME_MQTT_SLOTS]; + std::atomic _slot_health_recovering[RUNTIME_MQTT_SLOTS]; + std::atomic _slot_health_ack_wait_secs[RUNTIME_MQTT_SLOTS]; // Pending on-connect status publish: set from the onConnect callback (which // runs on the esp-mqtt event task, NOT this bridge task), consumed by the MQTT @@ -509,6 +521,9 @@ private: bool canStartSlotConnection(int index) const; void teardownSlot(int index, bool force = false); esp_err_t reconnectSlotClient(int index); + bool recoverSlotTransport(int index, uint32_t now); + void serviceSlotTransportHealth(uint32_t now); + void updateSlotTransportDiag(int index, uint32_t now); void maintainSlotConnections(); // Maintain all slot connections (token renewal, reconnect) void maintainSlotConnection(int index, unsigned long now_millis, unsigned long current_time, bool time_synced, bool& reconnect_attempted, bool& teardown_attempted); bool createSlotAuthToken(int index); // Create/renew JWT token for a slot diff --git a/test/README.md b/test/README.md index 1d4ff0201..8f2666f05 100644 --- a/test/README.md +++ b/test/README.md @@ -167,6 +167,7 @@ does not reflect the GoogleTest count -- run the built binary directly | `test_topic_template` | `src/helpers/MQTTTopicTemplate.h` | `{iata}/{device}/{token}/{type}` expansion, overflow/NUL-termination, and a buffer-size fuzz | | `test_mqtt_topic_router` | `src/helpers/MQTTTopicRouter.h` | complete preset/custom topic-routing contract; MeshRank all types except raw; required identifiers; invalid inputs/slots; exact buffer boundaries | | `test_mqtt_connection_policy` | `src/helpers/MQTTConnectionPolicy.h` | reconnect guard/backoff/stagger and breaker transitions; stable reset; JWT lifetime/renewal policy; exact timing boundaries and 32-bit `millis()` rollover | +| `test_mqtt_transport_health.py`, `test_mqtt_handshake_scheduler.py` | `src/helpers/MQTTTransportHealth.h`, production `MQTTBridge` methods | oldest outstanding QoS 1 acknowledgment deadline; quiet-session safety; lost/error-only callbacks; serialized stop-before-reconnect recovery; multi-hour scheduling and timer rollover | | `test_wifi_reconnect_policy` | `src/helpers/WiFiReconnectPolicy.h` | five-minute forced reconnect cadence, connection resets, duplicate disconnect observations, and 32-bit `millis()` rollover | | `test_radio_liveness` | `src/helpers/RadioLivenessTracker.h`, `src/helpers/radiolib/RxBoostedGainDefaults.h` | staged radio recovery, activity reset and rollover handling; target-specific RX boosted-gain defaults and SX126x precedence | | `test_lr1110_rx_recovery` | `src/helpers/radiolib/LR1110RxRecovery.h` | LR1110 four-byte RX-buffer shift signature; captured and accumulated shifts; ordinary/scoped packet exclusions | diff --git a/test/fixtures/mqtt_ap_ownership.cpp b/test/fixtures/mqtt_ap_ownership.cpp index 5f0a40020..08a2456cb 100644 --- a/test/fixtures/mqtt_ap_ownership.cpp +++ b/test/fixtures/mqtt_ap_ownership.cpp @@ -6,6 +6,7 @@ #include #include "helpers/AlertFaultPolicy.h" #include "helpers/MQTTConnectionPolicy.h" +#include "helpers/MQTTTransportHealth.h" #include "helpers/WiFiPowerSave.h" #include "helpers/WiFiNetworkPowerPolicy.h" #define MQTT_DEBUG_PRINTLN(...) ((void)0) @@ -90,8 +91,16 @@ struct MQTTBridge{ struct{bool canonical_wifi=true;}_node_info; struct{uint8_t wifi_power_save=1;}prefs,*_obs=&prefs; struct Client{void disconnect(){}bool isStarted()const{return false;}}; - struct Slot{Client* client=nullptr;bool connected=false;}_slots[RUNTIME_MQTT_SLOTS]; + struct Slot{ + Client* client=nullptr;bool connected=false; + uint32_t connected_at_ms=0,current_outage_started_ms=0; + MQTTTransportHealth::State transport_health; + }_slots[RUNTIME_MQTT_SLOTS]; bool _slot_attempt_pending[RUNTIME_MQTT_SLOTS]={}; + std::atomic _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS]={}; + std::atomic _slot_health_reset_pending[RUNTIME_MQTT_SLOTS]={}; + void updateSlotTransportDiag(int,uint32_t){} + void updateCachedConnectionStatus(){} std::vector cleanup; void cancelNtpRefresh(){cleanup.push_back("ntp");} void teardownSlot(int index){cleanup.push_back("slot"+std::to_string(index));} diff --git a/test/test_mqtt_connection_admission.py b/test/test_mqtt_connection_admission.py index 16db5ebc1..e47334027 100644 --- a/test/test_mqtt_connection_admission.py +++ b/test/test_mqtt_connection_admission.py @@ -15,6 +15,7 @@ POLICY_PREAMBLE = r''' #include #include #include "helpers/MQTTConnectionAdmission.h" +#include "helpers/MQTTTransportHealth.h" namespace Admission = MQTTConnectionAdmission; using MemoryState = Admission::ClientMemoryState; ''' @@ -65,8 +66,10 @@ struct MQTTBridge { const Preset* preset = nullptr; char broker_uri[128] = {}, host[64] = "broker"; uint16_t port = 8883; + MQTTTransportHealth::State transport_health; } _slots[RUNTIME_MQTT_SLOTS]; std::atomic _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {}; bool hasPendingSlotConnection() const; bool canStartSlotConnection(int index) const; }; diff --git a/test/test_mqtt_disconnect.py b/test/test_mqtt_disconnect.py index 10fbbc7c3..a0ad635fe 100644 --- a/test/test_mqtt_disconnect.py +++ b/test/test_mqtt_disconnect.py @@ -27,6 +27,7 @@ HARNESS = r''' #include #include #include +#include #define private public #include "PsychicMqttClient.h" #undef private @@ -100,6 +101,7 @@ struct MQTTBridge { PsychicMqttClient* client = nullptr; bool enabled = true, connected = false, circuit_breaker_tripped = false; uint32_t connected_at_ms = 0, last_reconnect_attempt = 0; + uint32_t current_outage_started_ms = 0; uint8_t reconnect_backoff = 0; bool initial_connect_done = true; char* auth_token = nullptr; @@ -107,11 +109,14 @@ struct MQTTBridge { uint32_t token_expires_at = 1, last_token_renewal = 1; uint32_t last_log_time = 1, last_deferred_log_ms = 1; uint8_t max_backoff_failures = 0; + MQTTTransportHealth::State transport_health; } _slots[RUNTIME_MQTT_SLOTS]; using MQTTSlot = Slot; mesh::wifi::NetworkPowerPolicy _network_power_policy; std::atomic _network_suspended{false}, _fresh_ntp_this_boot{false}; bool _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_reset_pending[RUNTIME_MQTT_SLOTS] = {}; bool _status_publish_pending[RUNTIME_MQTT_SLOTS] = {}; bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS] = {}; uint32_t _network_publish_acks[RUNTIME_MQTT_SLOTS] = {}; @@ -127,6 +132,7 @@ struct MQTTBridge { Board* _board = nullptr; void cancelNtpRefresh() {} void updateCachedConnectionStatus() {} + void updateSlotTransportDiag(int, uint32_t) {} void initializeWiFiInTask() {} void processPacketQueue() { ++queues; } void teardownSlot(int index, bool force = false); diff --git a/test/test_mqtt_handshake_scheduler.py b/test/test_mqtt_handshake_scheduler.py index f097624a4..653582196 100644 --- a/test/test_mqtt_handshake_scheduler.py +++ b/test/test_mqtt_handshake_scheduler.py @@ -21,11 +21,14 @@ PREAMBLE = r''' #include #include #include +#include +#include #include #include "helpers/AlertFaultPolicy.h" #include "helpers/MQTTConnectionAdmission.h" #include "helpers/MQTTConnectionHealth.h" #include "helpers/MQTTConnectionPolicy.h" +#include "helpers/MQTTTransportHealth.h" #include "helpers/WiFiPowerSave.h" #define ESP_PLATFORM 1 #define MQTT_DEBUG_PRINTLN(...) ((void)0) @@ -43,6 +46,11 @@ bool enforceStationChannel() { return true; } bool mqttStationWiFiMutationAllowed() { return true; } void esp_wifi_set_ps(wifi_ps_type_t) {} constexpr int MQTT_AUTH_JWT = 1; +constexpr int MQTT_ERROR_TYPE_CONNECTION_REFUSED = 5; +struct esp_mqtt_error_codes { + int esp_tls_last_esp_err = 0, esp_tls_stack_err = 0; + int esp_transport_sock_errno = 0, error_type = 0, connect_return_code = 0; +}; constexpr uint32_t MALLOC_CAP_INTERNAL = 1, MALLOC_CAP_DMA = 2; static uint32_t clock_ms = 1000; static unsigned long wall_seconds = 1800000000; @@ -59,7 +67,8 @@ struct { } WiFi; struct { size_t size = 2097152; size_t getPsramSize() const { return size; } } ESP; size_t heap_caps_get_free_size(uint32_t caps) { - assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); return dma_free; + assert(caps == MALLOC_CAP_INTERNAL || caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); + return dma_free; } size_t heap_caps_get_largest_free_block(uint32_t caps) { assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); return dma_largest; @@ -70,17 +79,61 @@ struct Preset { int auth_type = 0; unsigned long token_lifetime = 3300; bool enforce_exp = false; + bool allow_retain = false; }; bool mqttPresetEnforcesTokenExp(const Preset* preset) { return preset && preset->enforce_exp; } +namespace MQTTMessageBuilder { +void formatIsoTimestampForMqtt(long, long, const char*, char* buffer, size_t size) { + snprintf(buffer, size, "2026-10-14T00:00:00+00:00"); +} +template +int buildStatusMessage(int, const char*, const char*, const char*, const char*, + const char*, const char*, const char*, const char*, + char* buffer, int size, Args...) { + assert(size > 2); + strcpy(buffer, "{}"); + return 2; +} +} +void formatRadioInfo(float, float, int, int, char* buffer, size_t size) { + snprintf(buffer, size, "909.5,500,5,5"); +} struct PsychicMqttClient { bool initialized = false, started = false, live = false, model_task_heap = false; + bool stop_succeeds = true; int starts = 0, reconnects = 0, bounces = 0, credentials = 0, stops = 0; + uint32_t acknowledgments = 0; + int publish_result = 0, publishes = 0; + bool immediate_ack = false; + std::function connected_callback, disconnected_callback; + std::function error_callback; std::function disconnected; bool isInitialized() const { return initialized; } bool isStarted() const { return started; } bool connected() const { return live; } + uint32_t getPublishAck() const { return acknowledgments; } + void setAutoReconnect(bool enabled) { assert(!enabled); } + void onConnect(std::function callback) { connected_callback = callback; } + void onDisconnect(std::function callback) { disconnected_callback = callback; } + void onError(std::function callback) { error_callback = callback; } + void errorWithoutDisconnect() { + live = false; // The real adapter does this before calling onError. + assert(error_callback); + error_callback(esp_mqtt_error_codes{}); + } + void connectEvent() { + live = true; + assert(connected_callback); + connected_callback(false); + } + int publish(const char*, int qos, bool, const char*, int, bool async) { + assert(async == (qos > 0)); + ++publishes; + if (immediate_ack && qos > 0 && publish_result >= 0) ++acknowledgments; + return publish_result; + } int connect() { ++starts; if (model_task_heap) { @@ -93,9 +146,11 @@ struct PsychicMqttClient { int reconnect() { ++reconnects; return ESP_OK; } void disconnect() { ++stops; + if (!stop_succeeds) return; if (model_task_heap && started) dma_free += 6536; started = live = false; if (disconnected) disconnected(); + if (disconnected_callback) disconnected_callback(false); } void softDisconnect() { ++bounces; live = false; @@ -117,12 +172,26 @@ struct MQTTBridge { uint8_t reconnect_backoff = 0, max_backoff_failures = 0; bool circuit_breaker_tripped = false; uint32_t start_failures = 0; + uint32_t connect_failures = 0, disconnect_count = 0; + unsigned long first_disconnect_time = 0, current_outage_started_ms = 0; + unsigned long last_log_time = 0; + int32_t last_tls_err = 0, last_tls_stack_err = 0; + int last_sock_errno = 0; + unsigned long last_error_time = 0; + MQTTTransportHealth::State transport_health; } _slots[RUNTIME_MQTT_SLOTS]; PsychicMqttClient clients[RUNTIME_MQTT_SLOTS]; char tokens[RUNTIME_MQTT_SLOTS][16] = {}; bool setup_succeeds[RUNTIME_MQTT_SLOTS] = {true, true, true, true, true, true}; bool ready[RUNTIME_MQTT_SLOTS] = {true, true, true, true, true, true}; std::atomic _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_reset_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_recoveries[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_reason[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_recovering[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_ack_wait_secs[RUNTIME_MQTT_SLOTS] = {}; + bool _status_publish_pending[RUNTIME_MQTT_SLOTS] = {}; std::atomic _stop_requested{false}, _ntp_synced{true}; bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS] = {}; void* _identity = this; @@ -131,7 +200,11 @@ struct MQTTBridge { wl_status_t _last_wifi_status = WL_CONNECTED; unsigned long _last_wifi_check = 0, _last_wifi_reconnect_attempt = 0; uint8_t _wifi_reconnect_backoff_attempt = 0, _wifi_power_save = 1; - struct { bool canonical_wifi = false; } _node_info; + struct { + bool canonical_wifi = false; + float *freq = nullptr, *bw = nullptr; + int *sf = nullptr, *cr = nullptr; + } _node_info; struct Observer { uint8_t wifi_power_save = 1; } observer; Observer* _obs = &observer; AlertFaultPolicy::OutageSnapshot _wifi_outage{}; @@ -142,6 +215,39 @@ struct MQTTBridge { void beginWiFiStation() {} const char* _jwt_username = "jwt"; static const unsigned long SLOT_SETUP_RETRY_INTERVAL = 60000; + static const unsigned long SLOT_LOG_INTERVAL = 60000; + static constexpr int STATUS_JSON_BUFFER_SIZE = 1024, MSG_STATUS = 0; + const char *_timezone = "UTC", *_origin = "node", *_device_id = "id"; + const char *_board_model = "V4", *_firmware_version = "test"; + int _json_scratch_doc = 0, _queue_count = 0; + char _json_scratch_buffer[STATUS_JSON_BUFFER_SIZE] = {}; + struct Board { int getBattMilliVolts() const { return 4000; } }; + struct Clock { uint32_t getMillis() const { return clock_ms; } }; + struct Dispatcher { + int getErrFlags() const { return 0; } + int getTotalAirTime() const { return 0; } + int getReceiveAirTime() const { return 0; } + int getNumSentFlood() const { return 0; } + int getNumSentDirect() const { return 0; } + int getNumRecvFlood() const { return 0; } + int getNumRecvDirect() const { return 0; } + }; + struct Radio { + int getNoiseFloor() const { return -120; } + int getPacketsRecvErrors() const { return 0; } + }; + Board* _board = nullptr; + Clock* _ms = nullptr; + Dispatcher* _dispatcher = nullptr; + Radio* _radio = nullptr; + void refreshOriginFromPrefs() {} + void getClientVersion(char* output, size_t size) { snprintf(output, size, "test"); } + int repeatStatus() const { return 0; } + int collectConnHealth() const { return 0; } + bool buildTopicForSlot(int, int, char* output, size_t size) { + snprintf(output, size, "status"); + return true; + } std::vector setups; bool isSlotReady(int index) { return ready[index]; } bool setupSlot(int index) { @@ -162,7 +268,14 @@ struct MQTTBridge { void complete(int index) { _slot_attempt_pending[index].store(false, std::memory_order_release); _slots[index].connected = clients[index].live = true; + if (_slots[index].client) _slots[index].client->live = true; + _slots[index].connected_at_ms = clock_ms; + _slot_health_reset_pending[index] = true; } + void updateCachedConnectionStatus() {} + bool ensureSlotClient(int index); + bool publishToSlot(int, const char*, const char*, size_t, bool, uint8_t); + void publishStatusToSlot(int); void only(int index) { for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) _slots[i].enabled = i == index; } @@ -173,6 +286,9 @@ struct MQTTBridge { esp_err_t reconnectSlotClient(int index); unsigned long slotTokenLifetime(int index) const; void maintainSlotConnections(); + bool recoverSlotTransport(int index, uint32_t now); + void serviceSlotTransportHealth(uint32_t now); + void updateSlotTransportDiag(int index, uint32_t now); bool handleWiFiConnection(unsigned long); void maintainSlotConnection(int, unsigned long, unsigned long, bool, bool&, bool&); }; @@ -188,25 +304,36 @@ class MqttHandshakeSchedulerTest(unittest.TestCase): "bool MQTTBridge::hasPendingSlotConnection() const", "bool MQTTBridge::canStartSlotConnection(int index) const", "esp_err_t MQTTBridge::reconnectSlotClient(int index)", + "bool MQTTBridge::recoverSlotTransport(", + "void MQTTBridge::serviceSlotTransportHealth(", + "void MQTTBridge::updateSlotTransportDiag(", + "bool MQTTBridge::ensureSlotClient(", + "bool MQTTBridge::publishToSlot(", + "void MQTTBridge::publishStatusToSlot(", "unsigned long MQTTBridge::slotTokenLifetime(int index) const", "bool MQTTBridge::handleWiFiConnection(unsigned long now)", "void MQTTBridge::maintainSlotConnections()", "void MQTTBridge::maintainSlotConnection(")) - def compile_run(self, main): + def compile_run(self, main, fixture=None, expect_rejection=False): compiler = shutil.which("g++") or shutil.which("clang++") self.assertIsNotNone(compiler, "a host C++ compiler is required") with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-scheduler-") as temporary: work = Path(temporary) source, binary = work / "scheduler.cpp", work / "scheduler" - source.write_text(self.fixture() + main, encoding="utf-8") + source.write_text((self.fixture() if fixture is None else fixture) + main, + encoding="utf-8") result = subprocess.run( [compiler, "-std=c++17", "-Wall", "-Wextra", "-I", str(ROOT / "src"), str(source), "-o", str(binary)], capture_output=True, text=True, timeout=60) self.assertEqual(result.returncode, 0, result.stderr) result = subprocess.run([str(binary)], cwd=work, capture_output=True, text=True, timeout=10) - self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + if expect_rejection: + self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertIn("Assertion", result.stderr) + else: + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) def test_first_start_is_immediate_and_delayed_callback_serializes_following_slots(self): self.compile_run(r''' @@ -220,7 +347,7 @@ int main() { // still pending. Ordinary maintenance must not launch another SDK request. clock_ms = 16000; bridge.maintainSlotConnections(); - clock_ms = 3600000; + clock_ms = 60000; bridge.maintainSlotConnections(); assert(bridge.setups == std::vector{0}); assert(bridge.clients[0].starts == 1 && bridge.clients[0].reconnects == 0); @@ -419,7 +546,7 @@ int main() { bridge.maintainSlotConnections(); assert(bridge.clients[0].starts == 1 && dma_free == 35840); assert(bridge.hasPendingSlotConnection()); - clock_ms += 3600000; + clock_ms += 60000; bridge.maintainSlotConnections(); assert(bridge.clients[1].starts == 0 && bridge.clients[2].starts == 0); bridge.complete(0); @@ -450,6 +577,371 @@ int main() { } ''') + def test_missing_handshake_callback_times_out_only_after_sdk_stop(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; + bridge.only(0); + clock_ms = 1000; + bridge.maintainSlotConnections(); + assert(bridge.clients[0].starts == 1 && bridge._slot_attempt_pending[0]); + clock_ms = 120999; + bridge.maintainSlotConnections(); + assert(bridge.clients[0].stops == 0 && bridge._slot_attempt_pending[0]); + clock_ms = 121000; + bridge.maintainSlotConnections(); + auto& slot = bridge._slots[0]; + assert(bridge.clients[0].stops == 1 && !bridge.clients[0].isStarted()); + assert(!bridge._slot_attempt_pending[0] && !bridge.hasPendingSlotConnection()); + assert(slot.transport_health.recoveries == 1 && slot.connect_failures == 1); + assert(slot.transport_health.last_recovery_reason == + MQTTTransportHealth::RecoveryReason::HandshakeTimeout); + assert(slot.last_reconnect_attempt == clock_ms); + // Cleanup doesn't start another SDK task in the same maintenance pass. + assert(bridge.clients[0].starts == 1); + clock_ms += 14999; + bridge.maintainSlotConnections(); + assert(bridge.clients[0].starts == 1); + ++clock_ms; + bridge.maintainSlotConnections(); + assert(bridge.clients[0].starts == 2 && bridge._slot_attempt_pending[0]); + assert(slot.reconnect_backoff == 1); +} +''') + + def test_error_callback_without_disconnect_defers_cleanup_and_preserves_reservation(self): + self.compile_run(r''' +int main() { + for (bool established : {false, true}) { + MQTTBridge bridge; + bridge.only(0); + clock_ms = 1000; + assert(bridge.ensureSlotClient(0)); + auto& slot = bridge._slots[0]; + slot.initial_connect_done = true; + auto* client = slot.client; + assert(bridge.reconnectSlotClient(0) == ESP_OK); + if (established) client->connectEvent(); + client->errorWithoutDisconnect(); + // The actual production callback latches work. It cannot stop its own + // esp-mqtt task or release a still-running handshake from that callback. + assert(client->stops == 0 && client->isStarted()); + assert(bridge._slot_transport_fault_pending[0]); + assert(bridge.hasPendingSlotConnection() && !bridge.canStartSlotConnection(1)); + client->stop_succeeds = false; + bridge.maintainSlotConnections(); + assert(client->stops == 1 && client->isStarted()); + assert(bridge._slot_transport_fault_pending[0]); + assert(bridge._slot_attempt_pending[0] == !established); + assert(bridge.hasPendingSlotConnection() && !bridge.canStartSlotConnection(1)); + assert(slot.transport_health.recoveries == 0 && client->starts == 1); + // A later successful stop proves ownership ended, then recovery may admit + // another handshake. Failure accounting remains exactly once. + client->stop_succeeds = true; + bridge.maintainSlotConnections(); + assert(client->stops == 2 && !client->isStarted()); + assert(!bridge.hasPendingSlotConnection() && !bridge._slot_transport_fault_pending[0]); + assert(slot.transport_health.recoveries == 1); + assert(slot.connect_failures == (established ? 0u : 1u)); + assert(slot.transport_health.last_recovery_reason == + MQTTTransportHealth::RecoveryReason::TransportError); + assert(client->starts == 1); // completed cleanup starts the 15-second gap + delete client; + slot.client = nullptr; + } +} +''') + + def test_failed_stop_after_lost_callback_blocks_every_new_handshake(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; + clock_ms = 1000; + bridge.maintainSlotConnections(); + bridge.clients[0].stop_succeeds = false; + clock_ms += MQTTTransportHealth::kAttemptTimeoutMs; + for (int pass = 0; pass < 3; ++pass) { + bridge.maintainSlotConnections(); + assert(bridge.clients[0].stops == pass + 1 && bridge.clients[0].isStarted()); + assert(bridge._slot_attempt_pending[0] && bridge.hasPendingSlotConnection()); + assert(bridge._slots[0].transport_health.recoveries == 0); + assert(bridge.setups == std::vector{0}); + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) + assert(!bridge.canStartSlotConnection(i)); + clock_ms += 15000; + } + bridge.clients[0].stop_succeeds = true; + bridge.maintainSlotConnections(); + assert(!bridge.clients[0].isStarted() && !bridge.hasPendingSlotConnection()); + assert(bridge._slots[0].transport_health.recoveries == 1); + assert(bridge.setups == std::vector{0}); +} +''') + + def test_three_brokers_run_beyond_three_hours_and_only_missing_ack_recovers(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; + clock_ms = 1000; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { + bridge._slots[i].enabled = i < 3; + if (i >= 3) continue; + auto& slot = bridge._slots[i]; + slot.client = &bridge.clients[i]; + slot.initial_connect_done = true; + slot.client->initialized = slot.client->started = true; + bridge.complete(i); + slot.client->immediate_ack = true; + } + // Two status publishers get real ACK-counter progress; the third is quiet + // (for example, its broker does not accept status topics). Neither condition + // is a transport stall, irrespective of connection age or packet activity. + for (int minute = 0; minute <= 180; ++minute) { + assert(bridge.publishToSlot(0, "status", "{}", 2, true, 1)); + assert(bridge.publishToSlot(1, "status", "{}", 2, true, 1)); + bridge.maintainSlotConnections(); + for (int i = 0; i < 3; ++i) assert(bridge.clients[i].stops == 0); + clock_ms += 60000; + wall_seconds += 60; + } + assert(clock_ms > 3 * 60 * 60 * 1000); + bridge.clients[0].immediate_ack = false; + const uint32_t stalled_at = clock_ms; + assert(bridge.publishToSlot(0, "status", "{}", 2, true, 1)); + for (uint32_t elapsed = 0; elapsed < 120000; elapsed += 30000) { + clock_ms = stalled_at + elapsed; + assert(bridge.publishToSlot(0, "status", "{}", 2, true, 1)); + assert(bridge.publishToSlot(1, "status", "{}", 2, true, 1)); + bridge.maintainSlotConnections(); + assert(bridge.clients[0].stops == 0); + assert(bridge._slots[0].transport_health.ack_started_ms == stalled_at); + } + clock_ms = stalled_at + 120000; + bridge.maintainSlotConnections(); + assert(bridge.clients[0].stops == 1 && !bridge._slots[0].connected); + assert(bridge._slots[0].transport_health.recoveries == 1); + assert(bridge._slots[0].transport_health.last_recovery_reason == + MQTTTransportHealth::RecoveryReason::AckTimeout); + for (int i = 1; i < 3; ++i) { + assert(bridge.clients[i].stops == 0 && bridge._slots[i].connected); + assert(bridge._slots[i].transport_health.recoveries == 0); + } + assert(bridge._slot_health_recoveries[0] == 1); + assert(!bridge._slot_health_recovering[0]); +} +''') + + def test_on_connect_status_watch_survives_maintenance_and_ack_clears_it(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; + bridge.only(0); clock_ms = 1000; + assert(bridge.ensureSlotClient(0)); + auto& slot = bridge._slots[0]; + slot.initial_connect_done = true; + assert(bridge.reconnectSlotClient(0) == ESP_OK); + slot.client->connectEvent(); + assert(bridge._slot_health_reset_pending[0] && bridge._status_publish_pending[0]); + // The real worker consumes an on-connect status before it calls maintenance. + bridge.publishStatusToSlot(0); + assert(slot.client->publishes == 1); + assert(!bridge._slot_health_reset_pending[0] && slot.transport_health.ack_pending); + bridge.maintainSlotConnections(); + assert(slot.transport_health.ack_pending); + clock_ms += 119999; + ++slot.client->acknowledgments; + bridge.maintainSlotConnections(); + assert(!slot.transport_health.ack_pending && slot.client->stops == 0); + assert(bridge._slot_health_ack_wait_secs[0] == UINT32_MAX); + slot.client->immediate_ack = true; + assert(bridge.publishToSlot(0, "status", "{}", 2, true, 1)); + assert(!slot.transport_health.ack_pending); // ACK arrived inside enqueue + clock_ms += 3600000; + bridge.maintainSlotConnections(); + assert(slot.client->stops == 0); + delete slot.client; + slot.client = nullptr; +} +''') + + def test_negative_qos0_write_without_disconnect_requests_owner_recovery(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; + bridge.only(0); clock_ms = 1000; + auto& slot = bridge._slots[0]; + slot.client = &bridge.clients[0]; + slot.client->initialized = slot.client->started = true; + slot.initial_connect_done = true; + bridge.complete(0); + slot.client->publish_result = -1; + assert(!bridge.publishToSlot(0, "packets", "{}", 2, false, 0)); + assert(slot.client->stops == 0 && slot.connected && slot.client->connected()); + assert(slot.transport_health.recovery_reason == + MQTTTransportHealth::RecoveryReason::PublishFailure); + assert(bridge.hasPendingSlotConnection()); + assert(!bridge.publishToSlot(0, "packets", "{}", 2, false, 0)); + assert(slot.client->publishes == 1); // don't repeatedly hit the broken socket + bridge.maintainSlotConnections(); + assert(slot.client->stops == 1 && !slot.connected && !slot.client->isStarted()); + assert(slot.transport_health.recoveries == 1); + assert(slot.transport_health.last_recovery_reason == + MQTTTransportHealth::RecoveryReason::PublishFailure); +} +''') + + def test_handshake_and_ack_recovery_are_millis_wrap_safe(self): + self.compile_run(r''' +int main() { + for (bool awaiting_ack : {false, true}) { + MQTTBridge bridge; + bridge.only(0); clock_ms = UINT32_MAX - 60000; + bridge.maintainSlotConnections(); + auto& slot = bridge._slots[0]; + if (awaiting_ack) { + bridge.complete(0); + assert(bridge.publishToSlot(0, "status", "{}", 2, true, 1)); + } + clock_ms += 119999; + bridge.maintainSlotConnections(); + assert(slot.client->stops == 0); + ++clock_ms; + bridge.maintainSlotConnections(); + assert(slot.client->stops == 1 && slot.transport_health.recoveries == 1); + assert(slot.transport_health.last_recovery_reason == (awaiting_ack + ? MQTTTransportHealth::RecoveryReason::AckTimeout + : MQTTTransportHealth::RecoveryReason::HandshakeTimeout)); + } +} +''') + + def test_recovery_preserves_backoff_and_restart_memory_admission(self): + self.compile_run(r''' +int main() { + MQTTBridge bridge; + bridge.only(0); clock_ms = 70000; + auto& slot = bridge._slots[0]; + slot.client = &bridge.clients[0]; + slot.client->initialized = slot.client->started = true; + slot.initial_connect_done = true; + bridge.complete(0); + slot.reconnect_backoff = 3; + slot.max_backoff_failures = 1; + bridge._slot_transport_fault_pending[0] = true; + bridge.maintainSlotConnections(); + assert(slot.client->stops == 1 && slot.reconnect_backoff == 3); + assert(slot.max_backoff_failures == 1 && slot.connect_failures == 0); + clock_ms += 120000; + dma_free = 24575; dma_largest = 20000; + bridge.maintainSlotConnections(); + assert(slot.client->starts == 0 && slot.reconnect_backoff == 3); + assert(slot.last_reconnect_attempt == 70000 && slot.start_failures == 0); + dma_free = 24576; dma_largest = 8191; + bridge.maintainSlotConnections(); + assert(slot.client->starts == 0 && slot.reconnect_backoff == 3); + dma_largest = 8192; + bridge.maintainSlotConnections(); + assert(slot.client->starts == 1 && bridge._slot_attempt_pending[0]); + assert(slot.reconnect_backoff == 4 && slot.max_backoff_failures == 1); +} +''') + + def test_late_connected_callback_cannot_release_failed_stop_reservation(self): + self.compile_run(r''' +int main() { + for (bool publish_before_maintenance : {false, true}) { + MQTTBridge bridge; + bridge.only(0); clock_ms = 1000; + assert(bridge.ensureSlotClient(0)); + auto& slot = bridge._slots[0]; + slot.initial_connect_done = true; + auto* client = slot.client; + assert(bridge.reconnectSlotClient(0) == ESP_OK); + client->connectEvent(); + client->errorWithoutDisconnect(); + client->stop_succeeds = false; + bridge.maintainSlotConnections(); + assert(client->isStarted() && client->stops == 1); + assert(slot.transport_health.recovery_reason != MQTTTransportHealth::RecoveryReason::None); + // An in-flight SDK event arrives after stop failed. CONNACK may clear the + // attempt bit and set a new-session latch, but it cannot prove task join. + client->connectEvent(); + assert(!bridge._slot_attempt_pending[0] && bridge._slot_health_reset_pending[0]); + assert(bridge.hasPendingSlotConnection() && !bridge.canStartSlotConnection(1)); + if (publish_before_maintenance) { + assert(!bridge.publishToSlot(0, "status", "{}", 2, true, 1)); + assert(client->publishes == 0 && bridge.hasPendingSlotConnection()); + } + bridge.maintainSlotConnections(); + assert(client->stops == 2 && client->isStarted()); + assert(bridge.hasPendingSlotConnection() && !bridge.canStartSlotConnection(1)); + assert(client->starts == 1 && bridge.setups.empty()); + assert(slot.transport_health.recoveries == 0); + client->stop_succeeds = true; + bridge.maintainSlotConnections(); + assert(!client->isStarted() && !bridge.hasPendingSlotConnection()); + assert(slot.transport_health.recoveries == 1 && client->starts == 1); + delete client; + slot.client = nullptr; + } +} +''') + + def test_wifi_loss_failed_stop_and_reconnect_cannot_release_sdk_reservation(self): + self.compile_run(r''' +int main() { + for (bool established : {false, true}) { + MQTTBridge bridge; + bridge.only(0); clock_ms = 70000; WiFi.state = WL_CONNECTED; + assert(bridge.ensureSlotClient(0)); + auto& slot = bridge._slots[0]; + slot.initial_connect_done = true; + auto* client = slot.client; + assert(bridge.reconnectSlotClient(0) == ESP_OK); + if (established) client->connectEvent(); + client->stop_succeeds = false; + WiFi.state = WL_DISCONNECTED; + assert(!bridge.handleWiFiConnection(clock_ms)); + assert(client->isStarted() && client->stops == 1 && !slot.connected); + assert(bridge._slot_transport_fault_pending[0] && bridge.hasPendingSlotConnection()); + assert(!bridge.canStartSlotConnection(1)); + clock_ms += 11000; + WiFi.state = WL_CONNECTED; + assert(bridge.handleWiFiConnection(clock_ms)); + client->connectEvent(); // late SDK callback after WiFi recovered + assert(bridge.hasPendingSlotConnection()); + bridge.maintainSlotConnections(); + assert(client->isStarted() && client->stops == 2 && !slot.connected); + assert(bridge.hasPendingSlotConnection() && !bridge.canStartSlotConnection(1)); + assert(client->starts == 1 && bridge.setups.empty()); + client->stop_succeeds = true; + bridge.maintainSlotConnections(); + assert(!client->isStarted() && !bridge.hasPendingSlotConnection()); + assert(slot.transport_health.recoveries == 1 && client->starts == 1); + delete client; + slot.client = nullptr; + } +} +''') + + def test_pre_fix_error_callback_without_owner_signal_fails_regression(self): + source = self.fixture() + old_callback = source.replace( + "_slot_transport_fault_pending[index].store(true, std::memory_order_release);", + "/* pre-fix onError had no owner-task recovery signal */", 1) + self.assertNotEqual(source, old_callback) + self.compile_run(r''' +int main() { + MQTTBridge bridge; + bridge.only(0); clock_ms = 1000; + assert(bridge.ensureSlotClient(0)); + auto* client = bridge._slots[0].client; + assert(bridge.reconnectSlotClient(0) == ESP_OK); + client->errorWithoutDisconnect(); + assert(bridge._slot_transport_fault_pending[0]); +} +''', fixture=old_callback, expect_rejection=True) + if __name__ == "__main__": unittest.main() diff --git a/test/test_mqtt_network_timeout.py b/test/test_mqtt_network_timeout.py index 0586b7252..5d3fc1a20 100644 --- a/test/test_mqtt_network_timeout.py +++ b/test/test_mqtt_network_timeout.py @@ -21,6 +21,7 @@ PREAMBLE = r''' #include #include "helpers/MQTTConnectionAdmission.h" #include "helpers/MQTTConnectionPolicy.h" +#include "helpers/MQTTTransportHealth.h" struct esp_mqtt_client_config_t { #if ESP_IDF_VERSION_MAJOR == 5 struct { int keepalive = 0, message_retransmit_timeout = 0; } session; @@ -98,6 +99,7 @@ int main() { RECONFIGURE_PREAMBLE = r''' #include "helpers/MQTTConnectionAdmission.h" #include "helpers/MQTTConnectionPolicy.h" +#include "helpers/MQTTTransportHealth.h" #define MQTT_DEBUG_PRINTLN(...) ((void)0) #define pdMS_TO_TICKS(value) (value) constexpr int RUNTIME_MQTT_SLOTS = 2, MQTT_AUTH_JWT = 1; @@ -147,8 +149,11 @@ struct MQTTBridge { unsigned long last_reconnect_attempt = 0, last_log_time = 0, last_deferred_log_ms = 0; uint8_t reconnect_backoff = 0, max_backoff_failures = 0; bool circuit_breaker_tripped = false; + MQTTTransportHealth::State transport_health; } _slots[RUNTIME_MQTT_SLOTS]; std::atomic _stop_requested{false}, _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_reset_pending[RUNTIME_MQTT_SLOTS] = {}; bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS] = {}; bool _slot_reconfigure_pending[RUNTIME_MQTT_SLOTS] = {}; bool _initialized = true; @@ -161,6 +166,7 @@ struct MQTTBridge { bool canStartSlotConnection(int) const; void optimizeMqttClientConfig(PsychicMqttClient*, bool); void teardownSlot(int, bool force = false); + void updateSlotTransportDiag(int, uint32_t) {} void applySlotPreset(int, const char*); bool setupSlot(int index) { auto& slot = _slots[index]; diff --git a/test/test_mqtt_publish_ack.py b/test/test_mqtt_publish_ack.py index ee59e4d7b..d37c0c682 100644 --- a/test/test_mqtt_publish_ack.py +++ b/test/test_mqtt_publish_ack.py @@ -79,6 +79,7 @@ PREAMBLE = r''' #undef private #include "helpers/MQTTReplyFormat.h" #include "helpers/MQTTConnectionHealth.h" +#include "helpers/MQTTTransportHealth.h" #include "helpers/MQTTPacketFilter.h" #include "helpers/bridges/MQTTErrorLabels.h" template void test_log(Args&&...) {} @@ -161,6 +162,14 @@ struct MQTTBridge { int last_sock_errno = 0; }; MQTTSlot _slots[RUNTIME_MQTT_SLOTS]; + std::atomic _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_recovering[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_recoveries[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_reason[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_ack_wait_secs[RUNTIME_MQTT_SLOTS] = {}; + MQTTBridge() { + for (auto& wait : _slot_health_ack_wait_secs) wait.store(UINT32_MAX); + } bool _initialized = true; bool ready = true; MQTTPrefs* _obs = nullptr; @@ -437,6 +446,49 @@ int main() { } ''') + def test_production_recovery_diagnostics_never_claim_no_errors(self): + self.compile_run(r''' +int main() { + PsychicMqttClient client; + MQTTBridge bridge; + bridge._slots[0].client = &client; + s_mqtt_bridge_instance = &bridge; + bridge._slot_health_reason[0] = static_cast( + MQTTTransportHealth::RecoveryReason::AckTimeout); + bridge._slot_health_recoveries[0] = 3; + bridge._slot_health_ack_wait_secs[0] = 120; + for (bool fault_latched : {false, true}) { + bridge._slot_health_recovering[0] = !fault_latched; + bridge._slot_transport_fault_pending[0] = fault_latched; + char reply[160]; + MQTTBridge::formatSlotDiagReply(reply, sizeof(reply), 0); + assert(strcmp(reply, "> mqtt1: recover, pub_ack:0, rec:3/puback, ack_wait:120s") == 0); + assert(strstr(reply, "no errors") == nullptr); + } + bridge._slot_health_recovering[0] = false; + bridge._slot_transport_fault_pending[0] = false; + char recovered[160]; + MQTTBridge::formatSlotDiagReply(recovered, sizeof(recovered), 0); + assert(strstr(recovered, "> mqtt1: ok, no errors") == recovered); + assert(strstr(recovered, "rec:3/puback") != nullptr); // retain history + + bridge._slot_health_recovering[0] = true; + bridge._slot_health_recoveries[0] = UINT32_MAX; + bridge._slot_health_ack_wait_secs[0] = UINT32_MAX - 1; + for (size_t capacity = 1; capacity <= 160; ++capacity) { + unsigned char bounded[168]; + memset(bounded, 0xA5, sizeof(bounded)); + char* reply = reinterpret_cast(bounded); + MQTTBridge::formatSlotDiagReply(reply, capacity, 0); + assert(memchr(reply, '\0', capacity) != nullptr); + for (size_t pos = capacity; pos < sizeof(bounded); ++pos) assert(bounded[pos] == 0xA5); + // Every appended numeric field is omitted whole when it cannot fit. + if (strstr(reply, "rec:")) assert(strstr(reply, "rec:4294967295/puback")); + if (strstr(reply, "ack_wait:")) assert(strstr(reply, "ack_wait:4294967294s")); + } +} +''') + if __name__ == "__main__": unittest.main() diff --git a/test/test_mqtt_transport_health.py b/test/test_mqtt_transport_health.py new file mode 100644 index 000000000..7748b9aa9 --- /dev/null +++ b/test/test_mqtt_transport_health.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python3 +"""Exercise bounded MQTT liveness clocks without an SDK or PlatformIO build.""" +from pathlib import Path +import shutil +import subprocess +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parents[1] + + +class MqttTransportHealthTest(unittest.TestCase): + def compile_run(self, main): + compiler = shutil.which("g++") or shutil.which("clang++") + self.assertIsNotNone(compiler, "a host C++ compiler is required") + source = ('#include \n#include \n#include \n' + '#include "helpers/MQTTTransportHealth.h"\n' + 'using namespace MQTTTransportHealth;\n') + main + with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-health-") as directory: + work = Path(directory) + path, binary = work / "health.cpp", work / "health" + path.write_text(source, encoding="ascii") + result = subprocess.run( + [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", + "-I", str(ROOT / "src"), str(path), "-o", str(binary)], + capture_output=True, text=True, timeout=60) + self.assertEqual(result.returncode, 0, result.stderr) + result = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_repeated_acceptance_cannot_postpone_oldest_missing_ack(self): + self.compile_run(r''' +int main() { + State state; + state.acceptedQos1(1000, 7); + for (uint32_t now = 11000; now < 121000; now += 10000) { + state.acceptedQos1(now, 7); + assert(state.ack_started_ms == 1000 && state.ack_snapshot == 7); + assert(!state.ackExpired(now)); + } + assert(state.ackExpired(121000)); +} +''') + + def test_ack_advance_clears_watch_and_new_enqueue_starts_a_new_deadline(self): + self.compile_run(r''' +int main() { + State state; + state.acceptedQos1(1000, 7); + state.observeAck(7); + assert(state.ack_pending); + state.observeAck(8); + assert(!state.ack_pending && !state.ackExpired(121000)); + state.acceptedQos1(300000, 8); + assert(state.ack_started_ms == 300000 && state.ack_snapshot == 8); + assert(!state.ackExpired(419999) && state.ackExpired(420000)); + // An ACK arriving between maintenance passes is recognized on enqueue. + state.acceptedQos1(450000, 9); + assert(state.ack_started_ms == 450000 && state.ack_snapshot == 9); +} +''') + + def test_immediate_ack_race_uses_snapshot_from_before_enqueue(self): + self.compile_run(r''' +int main() { + State state; + const uint32_t before_enqueue = 11; + const uint32_t after_enqueue = 12; // broker event arrived inside SDK enqueue + state.acceptedQos1(1000, before_enqueue); + state.observeAck(after_enqueue); + assert(!state.ack_pending && !state.ackExpired(121000)); +} +''') + + def test_zero_start_ticks_and_millis_wrap_preserve_bounded_deadlines(self): + self.compile_run(r''' +int main() { + for (uint32_t start : {uint32_t(0), UINT32_MAX - 60000}) { + State state; + state.beginAttempt(start); + state.acceptedQos1(start, UINT32_MAX); + assert(!state.attemptExpired(start + kAttemptTimeoutMs - 1)); + assert(state.attemptExpired(start + kAttemptTimeoutMs)); + assert(!state.ackExpired(start + kAckTimeoutMs - 1)); + assert(state.ackExpired(start + kAckTimeoutMs)); + // The acknowledgment counter may wrap too; inequality still proves progress. + state.observeAck(0); + assert(!state.ack_pending); + state.completeAttempt(); + assert(!state.attemptExpired(start + kAttemptTimeoutMs)); + } +} +''') + + def test_recovery_keeps_first_reason_and_saturates_lifetime_counter(self): + self.compile_run(r''' +int main() { + State state; + state.beginAttempt(1000); + state.acceptedQos1(1000, 3); + state.requestRecovery(RecoveryReason::TransportError); + state.requestRecovery(RecoveryReason::AckTimeout); + assert(state.recovery_reason == RecoveryReason::TransportError); + state.recovered(); + assert(state.recoveries == 1 && !state.attempt_active && !state.ack_pending); + assert(state.recovery_reason == RecoveryReason::None); + assert(state.last_recovery_reason == RecoveryReason::TransportError); + state.recoveries = UINT32_MAX; + state.requestRecovery(RecoveryReason::PublishFailure); + state.recovered(); + assert(state.recoveries == UINT32_MAX); + assert(state.last_recovery_reason == RecoveryReason::PublishFailure); + assert(strcmp(reasonLabel(RecoveryReason::HandshakeTimeout), "handshake") == 0); + assert(strcmp(reasonLabel(RecoveryReason::AckTimeout), "puback") == 0); +} +''') + + def test_quiet_session_without_accepted_qos1_never_times_out(self): + self.compile_run(r''' +int main() { + State state; + state.beginAttempt(1000); + state.completeAttempt(); + for (uint32_t now = 0; now <= 7 * 24 * 60 * 60 * 1000; now += 60000) { + state.observeAck(0); + assert(!state.attemptExpired(now) && !state.ackExpired(now)); + assert(state.recovery_reason == RecoveryReason::None); + } +} +''') + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_mqtt_transport_results.py b/test/test_mqtt_transport_results.py index 2068783eb..c35a50305 100644 --- a/test/test_mqtt_transport_results.py +++ b/test/test_mqtt_transport_results.py @@ -32,6 +32,7 @@ PREAMBLE = r''' #include #include #include "helpers/MQTTConnectionHealth.h" +#include "helpers/MQTTTransportHealth.h" using esp_err_t = int; using esp_mqtt_client_handle_t = void*; constexpr int ESP_OK=0, ESP_FAIL=-1, ESP_ERR_INVALID_STATE=2, ESP_ERR_NO_MEM=3; @@ -53,6 +54,7 @@ static int register_result=ESP_OK, config_result=ESP_OK; static int start_result=ESP_OK, reconnect_result=ESP_OK; static int init_calls=0, register_calls=0, destroy_calls=0, config_calls=0; static int start_calls=0, reconnect_calls=0; +unsigned long millis() { return 1000; } static std::function sdk_callback; void* test_malloc(size_t size) { return fail_malloc ? nullptr : std::malloc(size); } #define malloc test_malloc @@ -88,9 +90,11 @@ struct MQTTBridge { struct MQTTSlot { PsychicMqttClient* client=nullptr; uint32_t start_failures=0; + MQTTTransportHealth::State transport_health; }; MQTTSlot _slots[RUNTIME_MQTT_SLOTS]; std::atomic _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {}; std::atomic _stop_requested{false}; bool hasPendingSlotConnection() const; bool canStartSlotConnection(int index) const; diff --git a/test/test_mqtt_wifi_network_power.py b/test/test_mqtt_wifi_network_power.py index 1a1672897..d3c23129f 100644 --- a/test/test_mqtt_wifi_network_power.py +++ b/test/test_mqtt_wifi_network_power.py @@ -22,6 +22,8 @@ BRIDGE = "src/helpers/bridges/MQTTBridge.cpp" PREAMBLE = r''' #include #include +#include +#include #include #include #include @@ -139,17 +141,18 @@ void configTime(int a, int b, const char* primary) { } struct Client { - bool started = true, fail_stop = false; + bool started = true, fail_stop = false, live = false; int stops = 0; uint32_t publish_acks = 0; uint32_t getPublishAck() const { return publish_acks; } bool isStarted() const { return started; } + bool connected() const { return live; } void disconnect() { require(!mesh::wifi::networkPower().backgroundSuspended(), "TLS stopped after handing station off"); ++stops; events.push_back("tls-stop"); - if (!fail_stop) started = false; + if (!fail_stop) started = live = false; if (on_client_stop) on_client_stop(); } }; @@ -185,9 +188,18 @@ struct MQTTBridge { uint32_t connected_at_ms = 0, last_reconnect_attempt = 200U; uint8_t reconnect_backoff = 5; Client* client = nullptr; + uint32_t connect_failures = 0, current_outage_started_ms = 0; + MQTTTransportHealth::State transport_health; } _slots[RUNTIME_MQTT_SLOTS]; + using MQTTSlot = Slot; Client clients[RUNTIME_MQTT_SLOTS]; std::atomic _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_reset_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_recovering[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_recoveries[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_reason[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _slot_health_ack_wait_secs[RUNTIME_MQTT_SLOTS] = {}; std::atomic _status_publish_pending[RUNTIME_MQTT_SLOTS] = {}; uint32_t _network_publish_acks[RUNTIME_MQTT_SLOTS] = {}; bool _network_espnow_restore_pending = false; @@ -237,6 +249,10 @@ struct MQTTBridge { bool isRunning() const { return true; } bool isNetworkSuspended() const { return _network_suspended.load(); } bool requestNtpTimeEstimate(); + bool hasPendingSlotConnection() const; + void serviceSlotTransportHealth(uint32_t now); + bool recoverSlotTransport(int index, uint32_t now); + void updateSlotTransportDiag(int index, uint32_t now); bool serviceNetworkPower(uint32_t now); void serviceEspNowChannelRestore(uint32_t now); void serviceBackgroundNtp(uint32_t now); @@ -336,6 +352,40 @@ static void failedSdkStopWithholdsHandoff() { "successful retry leaked a transport or radio"); } +static void failedSuspendLateConnectCannotAbandonTransportOwnership() { + resetEnvironment(); + MQTTBridge bridge; + for (auto& pending : bridge._slot_attempt_pending) pending = false; + bridge.serviceNetworkPower(clock_ms); + bridge.clients[1].fail_stop = true; + clock_ms += grace; + require(!bridge.serviceNetworkPower(clock_ms), "failed stop did not hold handoff"); + require(bridge._slot_transport_fault_pending[1] && bridge.hasPendingSlotConnection(), + "failed suspension abandoned a running MQTT transport"); + // A foreground owner wakes networking and a late CONNACK sets a fresh-session + // latch. Neither proves the failed SDK stop completed or frees admission. + mesh::wifi::networkPower().setHold(mesh::wifi::NetworkPowerHold::WebConfig, true); + bridge._slots[1].connected = bridge.clients[1].live = true; + bridge._slot_attempt_pending[1] = false; + bridge._slot_health_reset_pending[1] = true; + require(bridge.serviceNetworkPower(++clock_ms), "foreground owner did not keep radio"); + bridge.serviceSlotTransportHealth(clock_ms); + require(bridge.clients[1].started && bridge.hasPendingSlotConnection(), + "foreground wake or late CONNACK released failed-stop reservation"); + require(!bridge._slots[1].connected && !bridge._network_suspended, + "failed transport was advertised healthy or handed station away"); + require(bridge._slots[1].transport_health.recoveries == 0, + "failed SDK stop was counted as recovered"); + bridge.clients[1].fail_stop = false; + bridge.serviceSlotTransportHealth(++clock_ms); + require(!bridge.clients[1].started && !bridge.hasPendingSlotConnection(), + "proven successful stop did not release ownership"); + require(bridge._slots[1].transport_health.recoveries == 1, + "successful owner cleanup was not counted"); + require(WiFi.offs == 0 && WiFi.disconnects == 0, + "transport recovery stopped foreground WiFi"); +} + static void recoveryHasOneBoundedWindow() { for (uint32_t start : {100U, UINT32_MAX - 10000U}) { resetEnvironment(); @@ -686,6 +736,7 @@ int main(int argc, char** argv) { try { if (name == "handoff") failureStopsBothServicesBeforeWifi(); else if (name == "failed-stop") failedSdkStopWithholdsHandoff(); + else if (name == "failed-stop-wake") failedSuspendLateConnectCannotAbandonTransportOwnership(); else if (name == "recovery") recoveryHasOneBoundedWindow(); else if (name == "owners") foregroundAndEspNowArePreserved(); else if (name == "background-estimate") automaticClockEstimateCannotWakeSuspendedNetwork(); @@ -769,6 +820,10 @@ class MqttWiFiNetworkPowerTests(unittest.TestCase): "void MQTTBridge::refreshNTP(", "void MQTTBridge::pollNtpRefresh(", "bool MQTTBridge::requestNtpTimeEstimate(", + "bool MQTTBridge::hasPendingSlotConnection(", + "void MQTTBridge::serviceSlotTransportHealth(", + "bool MQTTBridge::recoverSlotTransport(", + "void MQTTBridge::updateSlotTransportDiag(", )) cls.temporary = tempfile.TemporaryDirectory(prefix="meshcore-mqtt-network-power-") cls.addClassCleanup(cls.temporary.cleanup) @@ -815,6 +870,9 @@ class MqttWiFiNetworkPowerTests(unittest.TestCase): def test_failed_sdk_transport_stop_keeps_station_owned_and_awake(self): self.run_case("failed-stop") + def test_failed_suspend_late_connect_and_foreground_wake_keep_sdk_reservation(self): + self.run_case("failed-stop-wake") + def test_recovery_restarts_once_and_stops_after_bounded_probe_across_wrap(self): self.run_case("recovery")