diff --git a/docs/cli_commands.md b/docs/cli_commands.md index 2e4edc2a9..f695ba6fc 100644 --- a/docs/cli_commands.md +++ b/docs/cli_commands.md @@ -4794,6 +4794,7 @@ MQTT preferences and retains explicit settings across reboots. - `set mqtt.enabled on` - `set mqtt.enabled off` - `get mqtt.running` +- `get mqtt.stopping` - `get mqtt.status` These commands are the same on MQTT-capable Companion, Repeater, Room Server, @@ -4803,6 +4804,14 @@ brokers to reconnect. `get mqtt.running` checks the service's runtime state; `get mqtt.status` reports individual broker connections. Enabling MQTT does not imply that a broker is connected. +On ESP32 Repeater and Room Server images, `set mqtt.enabled off` acknowledges +the stop request while the MQTT task disconnects and destroys its clients. +`get mqtt.stopping` stays `on` until that cleanup completes; `mqtt.running` +also stays `on` while those resources are still owned. Wait for +`mqtt.stopping` to become `off` before enabling MQTT again. UART, ESP-NOW, +and the radio continue running during this cleanup. OTA still waits for a +completed MQTT shutdown before it can write firmware. + USB logging remains independent. On images with both outputs, `set logging.output off|usb|wifi|both` selects USB and MQTT together. diff --git a/examples/simple_repeater/MyMesh.cpp b/examples/simple_repeater/MyMesh.cpp index decc12f6a..b5cefdbdd 100644 --- a/examples/simple_repeater/MyMesh.cpp +++ b/examples/simple_repeater/MyMesh.cpp @@ -12860,8 +12860,12 @@ void __attribute__((noinline)) MyMesh::servicePostMeshLoop() { expireRecentRepeatersIfDue(); #endif +#if defined(WITH_MQTT_BRIDGE) + // MQTT owns TLS on Core 0; the radio loop only reaps an acknowledged stop. + if (mqtt_bridge && mqtt_bridge->isRunning()) mqtt_bridge->loop(); +#endif #if defined(WITH_ESPNOW_BRIDGE) - // MQTT runs on Core 0. ESP-NOW remains cooperative, including in the + // ESP-NOW remains cooperative, including in the // combined Full image where both transports share the WiFi station radio. #if defined(WITH_MQTT_BRIDGE) || defined(WITH_RS232_BRIDGE) if (espnow_bridge.isRunning()) espnow_bridge.loop(); diff --git a/examples/simple_repeater/MyMesh.h b/examples/simple_repeater/MyMesh.h index e4a6efbc0..88fb0833d 100644 --- a/examples/simple_repeater/MyMesh.h +++ b/examples/simple_repeater/MyMesh.h @@ -1348,6 +1348,7 @@ public: _alerter.setBridge(nullptr); return !mqtt_bridge || !mqtt_bridge->isRunning(); } + if (mqtt_bridge && mqtt_bridge->isStopping()) return false; // Give the WiFi station to MQTT while it associates. ESP-NOW is restarted // after the AP channel is known, which avoids a raw ESP-NOW-only session // pinning the station to a different channel. @@ -1430,6 +1431,9 @@ public: #endif bool setBridgeState(bool enable) override { +#ifdef WITH_MQTT_BRIDGE + if (enable && mqtt_bridge && mqtt_bridge->isStopping()) return false; +#endif #if defined(WITH_MQTT_BRIDGE) && defined(WITH_ESPNOW_BRIDGE) if (!enable) { const bool espnow_stopped = setEspNowBridgeState(false); @@ -1665,6 +1669,24 @@ public: #endif } + bool requestMqttBridgeStop() override { +#ifdef WITH_MQTT_BRIDGE + _alerter.setBridge(nullptr); + if (mqtt_bridge) mqtt_bridge->requestStop(); + return true; +#else + return false; +#endif + } + + bool isMqttBridgeStopping() override { +#ifdef WITH_MQTT_BRIDGE + return mqtt_bridge && mqtt_bridge->isStopping(); +#else + return false; +#endif + } + bool syncMqttNtp() override { #ifdef WITH_MQTT_BRIDGE if (!mqtt_bridge || !mqtt_bridge->isRunning()) return false; diff --git a/examples/simple_room_server/MyMesh.cpp b/examples/simple_room_server/MyMesh.cpp index d85a27413..a5390f521 100644 --- a/examples/simple_room_server/MyMesh.cpp +++ b/examples/simple_room_server/MyMesh.cpp @@ -2604,7 +2604,8 @@ void MyMesh::loop() { sampleTelemetryHistory(); #endif #ifdef WITH_MQTT_BRIDGE - // bridge.loop() is now handled by FreeRTOS task on Core 0 - no need to call it here + // TLS runs on Core 0; the radio loop reaps only an acknowledged CLI stop. + if (bridge && bridge->isRunning()) bridge->loop(); #endif #ifdef WITH_ESPNOW_BRIDGE if (espnow_bridge.isRunning()) espnow_bridge.loop(); diff --git a/examples/simple_room_server/MyMesh.h b/examples/simple_room_server/MyMesh.h index 3d21f01f0..dc9afe4f6 100644 --- a/examples/simple_room_server/MyMesh.h +++ b/examples/simple_room_server/MyMesh.h @@ -610,6 +610,7 @@ public: _alerter.setBridge(nullptr); return !bridge || !bridge->isRunning(); } + if (bridge && bridge->isStopping()) return false; // Give the WiFi station to MQTT while it associates. ESP-NOW is restarted // after the AP channel is known, which avoids a raw ESP-NOW-only session // pinning the station to a different channel. @@ -702,6 +703,9 @@ public: } bool setBridgeState(bool enable) override { +#ifdef WITH_MQTT_BRIDGE + if (enable && bridge && bridge->isStopping()) return false; +#endif #if defined(WITH_MQTT_BRIDGE) && defined(WITH_ESPNOW_BRIDGE) if (!enable) return setEspNowBridgeState(false) && setMqttBridgeState(false); const bool mqtt_ok = _prefs.bridge_enabled @@ -846,6 +850,16 @@ public: return bridge && bridge->isRunning(); } + bool requestMqttBridgeStop() override { + _alerter.setBridge(nullptr); + if (bridge) bridge->requestStop(); + return true; + } + + bool isMqttBridgeStopping() override { + return bridge && bridge->isStopping(); + } + bool syncMqttNtp() override { if (!bridge || !bridge->isRunning()) return false; // Queue the sync onto the MQTT task (Core 0) without blocking: this runs on diff --git a/src/helpers/CommonCLI.cpp b/src/helpers/CommonCLI.cpp index e615a5856..c0a0fcb06 100644 --- a/src/helpers/CommonCLI.cpp +++ b/src/helpers/CommonCLI.cpp @@ -3786,6 +3786,10 @@ void CommonCLI::handleSetCmd(uint32_t sender_timestamp, char* command, char* rep strcpy(reply, "Error: usage set logging.output off|usb|wifi|both"); return; } + if (wifi_enabled && _callbacks->isMqttBridgeStopping()) { + strcpy(reply, "Error: MQTT is stopping; retry when mqtt.stopping is off"); + return; + } const uint8_t previous_usb = _prefs->usb_logging_enabled; const uint8_t previous_bridge = _prefs->bridge_enabled; _prefs->usb_logging_enabled = usb_enabled ? 1 : 0; @@ -3797,7 +3801,9 @@ void CommonCLI::handleSetCmd(uint32_t sender_timestamp, char* command, char* rep return; } mesh::setUsbLoggingEnabled(usb_enabled); - if (!_callbacks->setMqttBridgeState(wifi_enabled)) { + const bool applied = wifi_enabled ? _callbacks->setMqttBridgeState(true) + : _callbacks->requestMqttBridgeStop(); + if (!applied) { strcpy(reply, "Error: logging output saved, but MQTT runtime change failed"); return; } @@ -3856,9 +3862,21 @@ void CommonCLI::handleSetCmd(uint32_t sender_timestamp, char* command, char* rep strcpy(reply, "Error: use set mqtt.enabled on|off"); } else { const bool enable = strcmp(value, "on") == 0; + if (enable && _callbacks->isMqttBridgeStopping()) { + strcpy(reply, "Error: MQTT is stopping; retry when mqtt.stopping is off"); + return; + } + const uint8_t previous = _prefs->bridge_enabled; _prefs->bridge_enabled = enable ? 1 : 0; - const bool applied = _callbacks->setMqttBridgeState(enable); - savePrefs(); + // A cooperative stop cannot be cancelled safely once signalled. Save + // intent first so a failed write leaves the running service unchanged. + if (!trySavePrefs()) { + _prefs->bridge_enabled = previous; + strcpy(reply, "Error: MQTT setting not saved; unchanged"); + return; + } + const bool applied = enable ? _callbacks->setMqttBridgeState(true) + : _callbacks->requestMqttBridgeStop(); strcpy(reply, applied ? "OK" : "Error: MQTT runtime change failed; setting saved"); } diff --git a/src/helpers/CommonCLI.h b/src/helpers/CommonCLI.h index 985971de8..ed67a03a6 100644 --- a/src/helpers/CommonCLI.h +++ b/src/helpers/CommonCLI.h @@ -603,6 +603,10 @@ public: // legacy bridge behavior by default. Combined bridge roles override // them to let each transport be started, stopped, and queried independently. virtual bool setMqttBridgeState(bool enable) { return setBridgeState(enable); } + // CLI disable may acknowledge before worker-owned TLS teardown finishes. + // Ownership barriers continue to use the synchronous state setter. + virtual bool requestMqttBridgeStop() { return setMqttBridgeState(false); } + virtual bool isMqttBridgeStopping() { return false; } virtual bool setEspNowBridgeState(bool enable) { return setBridgeState(enable); } virtual bool setRs232BridgeState(bool enable) { return setBridgeState(enable); } virtual bool restartMqttBridge() { return restartBridge(); } diff --git a/src/helpers/CommonCLI_Observer.cpp b/src/helpers/CommonCLI_Observer.cpp index ce3d3b8de..d97fe55ef 100644 --- a/src/helpers/CommonCLI_Observer.cpp +++ b/src/helpers/CommonCLI_Observer.cpp @@ -880,6 +880,10 @@ bool CommonCLI::handleObserverGetCmd(uint32_t sender_timestamp, const char* conf snprintf(reply, 160, "> %s", _callbacks->isMqttBridgeRunning() ? "on" : "off"); return true; } + if (strcmp(config, "mqtt.stopping") == 0) { + snprintf(reply, 160, "> %s", _callbacks->isMqttBridgeStopping() ? "on" : "off"); + return true; + } if (memcmp(config, "snmp.community", 14) == 0) { sprintf(reply, "> %s", _mqtt_prefs.snmp_community); } else if (memcmp(config, "snmp", 4) == 0 && (config[4] == '\0' || config[4] == '\n' || config[4] == '\r')) { diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 4cd401f1b..d01167539 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -683,7 +683,7 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) // end() returns as soon as the task acks (it checks _stop_acked before ticking // the timeout), so a larger bound does NOT slow a healthy stop -- it only length- // ens the wait before force-killing a genuinely wedged task. The timeout is set -// per stop in end() via computeStopTimeoutMs() based on the enabled-slot count. +// per stop in requestStop() based on the enabled-slot count. static const uint32_t MQTT_STOP_TIMEOUT_BASE_MS = 5000; // fixed teardown overhead static const uint32_t MQTT_STOP_TIMEOUT_PER_SLOT_MS = 8000; // ~5-6 s measured + headroom @@ -899,8 +899,8 @@ void MQTTBridge::begin() { // Idempotent start: a second begin() would re-run allocation and re-create // the task, leaking the previous queue and task. - if (_initialized) { - MQTT_DEBUG_PRINTLN("MQTT Bridge already running - begin() ignored"); + if (_initialized || _lifecycle.isStopInProgress()) { + MQTT_DEBUG_PRINTLN("MQTT Bridge running or stopping - begin() ignored"); return; } @@ -1078,6 +1078,10 @@ void MQTTBridge::begin() { } } + // Reset the handshake before starting either the worker or cooperative loop. + _stop_requested.store(false, std::memory_order_release); + _stop_acked.store(false, std::memory_order_release); + #ifdef ESP_PLATFORM // Create FreeRTOS queue; use PSRAM storage when available #ifdef BOARD_HAS_PSRAM @@ -1121,11 +1125,6 @@ void MQTTBridge::begin() { // Task stack: dynamic allocation (internal RAM). A PSRAM-backed stack was tried and // reverted -- it resets some boards (e.g. Heltec V4) when the task runs from PSRAM. _mqtt_task_handle = nullptr; - // Clear the cooperative-stop handshake before the new task starts reading it. - // deliverStop() leaves _stop_requested latched true after a stop cycle, so a - // restart must reset it or the fresh task would self-terminate immediately. - _stop_requested = false; - _stop_acked = false; BaseType_t create_result = xTaskCreatePinnedToCore( mqttTask, "MQTTBridge", @@ -1179,16 +1178,11 @@ void MQTTBridge::begin() { } // --------------------------------------------------------------------------- -// end() +// requestStop() - signal only; worker retains ownership until acknowledgment // --------------------------------------------------------------------------- -void MQTTBridge::end() { - MQTT_DEBUG_PRINTLN("Stopping MQTT Bridge..."); - - // Idempotent stop: nothing to tear down if we never started (or already stopped). - if (!_initialized) { - MQTT_DEBUG_PRINTLN("MQTT Bridge already stopped - end() ignored"); - return; - } +void MQTTBridge::requestStop() { + // A repeated request must not clear an acknowledgment or restart its budget. + if (!_initialized || _lifecycle.isStopInProgress()) return; // Stop new diagnostic reads through the singleton before teardown begins. s_mqtt_bridge_instance = nullptr; @@ -1216,12 +1210,28 @@ void MQTTBridge::end() { // task mid-mbedTLS and then free client buffers on a corrupted heap. _lifecycle.requestStop(); // Running -> StopRequested; deliverStop() sets _stop_requested +#ifndef ESP_PLATFORM + // The cooperative implementation has no worker to acknowledge separately. + _stop_acked.store(true, std::memory_order_release); + _lifecycle.onTaskStopped(); + finishStopped(); +#endif +} + +// --------------------------------------------------------------------------- +// end() - synchronous barrier for OTA and other ownership transfers +// --------------------------------------------------------------------------- +void MQTTBridge::end() { + MQTT_DEBUG_PRINTLN("Stopping MQTT Bridge..."); + if (!_initialized) return; + requestStop(); // Joins an existing request without re-delivering its signal. + #ifdef ESP_PLATFORM // Wait (bounded) for the task to acknowledge. tick() synthesizes the timeout // fallback if the task never acks. Checking the ack first each iteration means // a stop that completes right as the timeout expires is still treated as clean. while (_lifecycle.isStopInProgress()) { - if (_stop_acked) { + if (_stop_acked.load(std::memory_order_acquire)) { _lifecycle.onTaskStopped(); // StopRequested -> Stopped (clean): releaseResources() break; } @@ -1229,14 +1239,12 @@ void MQTTBridge::end() { if (!_lifecycle.isStopInProgress()) break; vTaskDelay(pdMS_TO_TICKS(20)); } -#else - // Non-ESP32: the bridge runs cooperatively in loop(); there is no separate - // task to signal. Drive straight to a clean Stopped and let releaseResources() - // perform the (unchanged) synchronous teardown. - _stop_acked = true; - _lifecycle.onTaskStopped(); #endif + finishStopped(); +} +void MQTTBridge::finishStopped() { + if (!_initialized || _lifecycle.isStopInProgress()) return; // Timezone is inline class storage (_timezone_storage) - nothing to delete. // The shared JSON document's pools were freed by releaseRuntimeBuffers() above. _initialized = false; @@ -1249,6 +1257,7 @@ void MQTTBridge::end() { _ntp_estimate_done = false; _ntp_estimate_ok = false; _ntp_estimate_epoch = 0; + _staged_raw_valid = false; MQTT_DEBUG_PRINTLN("MQTT Bridge stopped (%s)", _lifecycle.stopTimedOut() ? "forced/timeout - OTA blocked" : "clean"); } @@ -1256,7 +1265,7 @@ void MQTTBridge::end() { // --------------------------------------------------------------------------- // LifecycleOps - binds MQTTLifecycle::Ops (the pure, host-tested spec) to the // FreeRTOS / PsychicMqttClient runtime. Every method runs on the loop task -// (Core 1): the Coordinator that calls them is driven only from begin()/end(). +// (Core 1): begin(), requestStop(), end(), and loop() own the Coordinator. // --------------------------------------------------------------------------- uint32_t MQTTBridge::LifecycleOps::nowMs() { return (uint32_t)millis(); @@ -1271,8 +1280,8 @@ void MQTTBridge::LifecycleOps::deliverStop() { // Clear any stale ack before raising the request (same ordering as the NTP // handshake: clear the done-flag, then set the request). The MQTT task polls // _stop_requested at the top of mqttTaskLoop(). - _b->_stop_acked = false; - _b->_stop_requested = true; + _b->_stop_acked.store(false, std::memory_order_release); + _b->_stop_requested.store(true, std::memory_order_release); } void MQTTBridge::LifecycleOps::releaseResources() { @@ -1426,6 +1435,23 @@ void MQTTBridge::initializeWiFiInTask() { MQTT_DEBUG_PRINTLN("WiFi initialization started in task"); } +// --------------------------------------------------------------------------- +// Preserve startup/TLS spacing while allowing the owning task to stop promptly. +// Unsigned tick subtraction remains valid across the FreeRTOS tick wrap. +bool MQTTBridge::waitUnlessStopping(uint32_t delay_ms) { + const TickType_t duration = pdMS_TO_TICKS(delay_ms); + const TickType_t started = xTaskGetTickCount(); + TickType_t quantum = pdMS_TO_TICKS(20); + if (quantum == 0) quantum = 1; + while (!_stop_requested) { + const TickType_t elapsed = xTaskGetTickCount() - started; + if (elapsed >= duration) return !_stop_requested; + const TickType_t remaining = duration - elapsed; + vTaskDelay(remaining < quantum ? remaining : quantum); + } + return false; +} + // --------------------------------------------------------------------------- // mqttTaskLoop() - main loop running on Core 0 // --------------------------------------------------------------------------- @@ -1434,7 +1460,7 @@ void MQTTBridge::mqttTaskLoop() { initializeWiFiInTask(); // Wait a bit for WiFi to start connecting - vTaskDelay(pdMS_TO_TICKS(1000)); + waitUnlessStopping(1000); // Main task loop #ifdef MQTT_MEMORY_DEBUG @@ -1454,7 +1480,7 @@ void MQTTBridge::mqttTaskLoop() { teardownSlot(i); } destroySlotClients(); - _stop_acked = true; // release semantics: set only after teardown is done + _stop_acked.store(true, std::memory_order_release); // teardown precedes Core 1 release return; } @@ -1556,6 +1582,10 @@ void MQTTBridge::mqttTaskLoop() { } #endif + // Network/time probes may have yielded while Core 1 requested a stop. + // Return to the owner-core teardown before allocating another client. + if (_stop_requested) continue; + // Deferred slot setup: wait until NTP is synced so JWT tokens get valid timestamps. // This avoids wasted TLS handshakes that get rejected due to bad token times. if (_ntp_synced.load(std::memory_order_acquire) && !_slots_setup_done) { @@ -1570,6 +1600,7 @@ void MQTTBridge::mqttTaskLoop() { MQTT_DEBUG_PRINTLN("NTP synced, setting up MQTT slots (max %d active)...", _max_active_slots); for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_stop_requested) break; if (_slots[i].enabled) { if (!canActivateSlot(i)) { MQTT_DEBUG_PRINTLN("MQTT%d skipped: max active slots (%d) reached", i + 1, _max_active_slots); @@ -1588,20 +1619,26 @@ void MQTTBridge::mqttTaskLoop() { // Stagger connections: 5s between slots to avoid simultaneous TLS handshakes // which compete for ~40KB internal heap each if (i < RUNTIME_MQTT_SLOTS - 1) { - vTaskDelay(pdMS_TO_TICKS(5000)); + if (!waitUnlessStopping(5000)) break; } } } } + // A stop during a stagger must not fall through into reconfiguration, + // publication, or connection maintenance before the next loop iteration. + if (_stop_requested) continue; + // Process pending slot reconfigures (queued from CLI on Core 1) for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_stop_requested) break; if (_slot_reconfigure_pending[i]) { _slot_reconfigure_pending[i] = false; MQTT_DEBUG_PRINTLN("Applying deferred reconfigure for MQTT%d (preset: %s)", i + 1, _obs->mqtt_slot_preset[i]); applySlotPreset(i, _obs->mqtt_slot_preset[i]); } } + if (_stop_requested) continue; // Publish on-connect status for slots whose onConnect callback fired since // the last loop. Raised on the esp-mqtt event task, consumed here on the @@ -1610,17 +1647,21 @@ void MQTTBridge::mqttTaskLoop() { // reconnect during the publish re-arms for the next loop rather than being // lost; publishStatusToSlot() re-checks slot.connected and no-ops if dropped. for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_stop_requested) break; if (_status_publish_pending[i]) { _status_publish_pending[i] = false; publishStatusToSlot(i); } } + if (_stop_requested) continue; // Maintain slot connections (token renewal, reconnect with backoff) maintainSlotConnections(); + if (_stop_requested) continue; // Process packet queue processPacketQueue(); + if (_stop_requested) continue; #if defined(WITH_MQTT_NEIGHBORS) // Consume a pending neighbors snapshot handed over by the mesh (Core 1). @@ -1939,6 +1980,7 @@ bool MQTTBridge::canActivateSlot(int index) const { // maintainSlotConnections() will retry it -- the allocation failures below are transient // memory conditions, not permanent misconfiguration. bool MQTTBridge::setupSlot(int index) { + if (_stop_requested.load(std::memory_order_acquire)) return false; if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; MQTTSlot& slot = _slots[index]; @@ -2209,6 +2251,7 @@ void MQTTBridge::teardownSlot(int index, bool force) { esp_err_t MQTTBridge::reconnectSlotClient(int index) { if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return ESP_ERR_INVALID_ARG; + if (_stop_requested.load(std::memory_order_acquire)) return ESP_ERR_INVALID_STATE; MQTTSlot& slot = _slots[index]; if (slot.client == nullptr) return ESP_ERR_INVALID_STATE; @@ -2234,6 +2277,7 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { } void MQTTBridge::maintainSlotConnections() { + if (_stop_requested.load(std::memory_order_acquire)) return; if (!_identity) return; // Check WiFi status first @@ -2271,6 +2315,7 @@ void MQTTBridge::maintainSlotConnections() { bool setup_retry_this_cycle = false; for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_stop_requested) return; if (!_slots[i].enabled) continue; // JWT slots need time sync before we can manage tokens @@ -2617,6 +2662,7 @@ bool MQTTBridge::createSlotAuthToken(int index) { } bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload, size_t payload_len, bool retained, uint8_t qos) { + if (_stop_requested.load(std::memory_order_acquire)) return false; if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; MQTTSlot& slot = _slots[index]; if (!slot.client || !slot.connected) { @@ -2854,6 +2900,7 @@ bool MQTTBridge::isAnySlotConnected() { } void MQTTBridge::setSlotPreset(int slot_index, const char* preset_name) { + if (_stop_requested.load(std::memory_order_acquire)) return; if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return; // On ESP32, teardown/setup involves TLS and must run on the MQTT task (Core 0). @@ -2871,6 +2918,7 @@ void MQTTBridge::setSlotPreset(int slot_index, const char* preset_name) { } void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) { + if (_stop_requested.load(std::memory_order_acquire)) return; if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return; MQTTSlot& slot = _slots[slot_index]; @@ -2943,6 +2991,7 @@ void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) { void MQTTBridge::setSlotCustomBroker(int slot_index, const char* host, uint16_t port, const char* username, const char* password) { + if (_stop_requested.load(std::memory_order_acquire)) return; if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return; MQTTSlot& slot = _slots[slot_index]; @@ -3085,7 +3134,8 @@ bool MQTTBridge::handleWiFiConnection(unsigned long now) { } bool MQTTBridge::isReady() const { - return _initialized && isWiFiConfigValid(_obs); + return _initialized && !_stop_requested.load(std::memory_order_acquire) + && isWiFiConfigValid(_obs); } bool MQTTBridge::isIATAValid() const { @@ -3135,13 +3185,20 @@ bool MQTTBridge::isSlotReady(int index, char* reason_buf, size_t reason_size) co } // --------------------------------------------------------------------------- -// loop() - non-ESP32 main loop (ESP32 uses mqttTaskLoop via FreeRTOS task) +// loop() - Core 1 stop reaping on ESP32; cooperative processing elsewhere // --------------------------------------------------------------------------- void MQTTBridge::loop() { if (!_initialized) return; #ifdef ESP_PLATFORM - // On ESP32, loop() is a no-op - all processing happens in the FreeRTOS task + // Poll only: a slow TLS teardown must not stall radio/UART forwarding or + // force-delete the worker. Only end(), the synchronous OTA barrier, owns + // the existing bounded-timeout fallback. + if (_lifecycle.isStopInProgress() + && _stop_acked.load(std::memory_order_acquire)) { + _lifecycle.onTaskStopped(); + finishStopped(); + } return; #else unsigned long now = millis(); @@ -3271,7 +3328,8 @@ void MQTTBridge::loop() { // --------------------------------------------------------------------------- void MQTTBridge::onPacketReceived(mesh::Packet *packet) { - if (!_initialized || !_obs->mqtt_packets_enabled || !_obs->mqtt_rx_enabled) return; + if (!_initialized || _stop_requested.load(std::memory_order_acquire) + || !_obs->mqtt_packets_enabled || !_obs->mqtt_rx_enabled) return; // Drop before the queue copy when no configured slot allows this payload // type. A QueuedPacket carries the packet plus up to 256 bytes of raw radio @@ -3295,7 +3353,8 @@ void MQTTBridge::onPacketReceived(mesh::Packet *packet) { void MQTTBridge::sendPacket(mesh::Packet *packet) { uint8_t tx_mode = _obs->mqtt_tx_enabled; // Read live from prefs (no restart needed) - if (!_initialized || !_obs->mqtt_packets_enabled || tx_mode == 0) return; + if (!_initialized || _stop_requested.load(std::memory_order_acquire) + || !_obs->mqtt_packets_enabled || tx_mode == 0) return; // Advert mode: only queue self-originated advert packets if (tx_mode == 2) { @@ -4000,7 +4059,7 @@ void MQTTBridge::requestPublishNeighbors(const char* json, size_t len) { if (_neighbors_publish_pending.load(std::memory_order_acquire)) return; // A discovery can finish after bridge shutdown. Do not allocate a buffer // which has no task left to consume or release it. - if (!isRunning()) return; + if (!isRunning() || _stop_requested.load(std::memory_order_acquire)) return; _neighbors_json_buffer = static_cast( MQTTRuntimeBufferLifecycle::allocateIfMissing( _neighbors_json_buffer, NEIGHBORS_JSON_BUFFER_SIZE, psram_malloc)); @@ -4047,6 +4106,7 @@ bool MQTTBridge::publishNeighbors() { // --------------------------------------------------------------------------- void MQTTBridge::queuePacket(mesh::Packet* packet, bool is_tx) { + if (!_initialized || _stop_requested.load(std::memory_order_acquire)) return; #ifdef ESP_PLATFORM // Use FreeRTOS queue for thread-safe operation if (_packet_queue_handle == nullptr) { @@ -4174,6 +4234,7 @@ void MQTTBridge::dequeuePacket() { // --------------------------------------------------------------------------- void MQTTBridge::storeRawRadioData(const uint8_t* raw_data, int len, float snr, float rssi) { + if (!_initialized || _stop_requested.load(std::memory_order_acquire)) return; // Writes into the Core 1-only staging area. No mutex needed: this function and // queuePacket() are both called from Core 1 in guaranteed sequence for each packet. if (len > 0 && len <= (int)LAST_RAW_DATA_SIZE) { @@ -4549,7 +4610,7 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) { } bool MQTTBridge::requestForcedNtpSync(uint32_t timeout_ms) { - if (!isRunning()) return false; + if (!isRunning() || _stop_requested.load(std::memory_order_acquire)) return false; // Publish the request to the MQTT task. Clear the completion flags before // raising _ntp_force_requested so the task can't observe a stale result. @@ -4618,7 +4679,8 @@ void MQTTBridge::runNtpEstimateProbe() { } bool MQTTBridge::requestNtpTimeEstimate() { - if (!isRunning() || WiFi.status() != WL_CONNECTED || _ntp_estimate_requested) return false; + if (!isRunning() || _stop_requested.load(std::memory_order_acquire) + || WiFi.status() != WL_CONNECTED || _ntp_estimate_requested) return false; _ntp_estimate_done = false; _ntp_estimate_ok = false; _ntp_estimate_epoch = 0; @@ -4636,7 +4698,8 @@ bool MQTTBridge::takeNtpTimeEstimate(uint32_t& epoch, bool& finished) { } bool MQTTBridge::ntpDiag(char* reply, size_t reply_size, bool verbose) { - if (!isRunning() || reply == nullptr || reply_size == 0) return false; + if (!isRunning() || _stop_requested.load(std::memory_order_acquire) + || reply == nullptr || reply_size == 0) return false; // Marshal the probe onto the MQTT task (Core 0); clear the completion flag first. _ntp_diag_done = false; diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 128d963c1..ab7c43ade 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -300,12 +300,11 @@ private: // Cooperative-shutdown handshake (Phase 5). The loop task (Core 1) raises // _stop_requested through the lifecycle Coordinator; the MQTT task (Core 0) // sees it, tears down its own clients on Core 0 (where the mbedTLS contexts - // live), sets _stop_acked LAST, and self-terminates. end() waits for the ack - // before freeing the queue/buffers. Plain volatile matches the existing - // NTP/reconfigure handshake idiom above; replacing all of these with a command - // channel / task notifications is explicitly deferred (see MQTT_OWNERSHIP.md). - volatile bool _stop_requested = false; - volatile bool _stop_acked = false; + // live), sets _stop_acked LAST, and self-terminates. end() waits for the ack; + // loop() can instead reap it without blocking. Atomic release/acquire ordering + // makes client teardown visible before Core 1 frees the queue/buffers. + std::atomic _stop_requested{false}; + std::atomic _stop_acked{false}; // Timezone handling. // _timezone_storage is inline class storage (zero heap) that is reconfigured @@ -515,6 +514,7 @@ private: #ifdef ESP_PLATFORM static void mqttTask(void* parameter); void mqttTaskLoop(); // Main loop for MQTT task + bool waitUnlessStopping(uint32_t delay_ms); void initializeWiFiInTask(); // WiFi initialization moved to task #endif bool publishPacket(mesh::Packet* packet, bool is_tx, bool& has_eligible_target, @@ -551,13 +551,14 @@ private: // of making the bridge unusable. void allocateRuntimeBuffers(); void releaseRuntimeBuffers(); + void finishStopped(); // Core 1 only, after lifecycle resource release // --- Cooperative lifecycle (Phase 5) --------------------------------------- // The pure state machine, bounded stop timeout, and OTA barrier live in // src/helpers/MQTTLifecycle.h and are host-tested by test/test_mqtt_lifecycle/. // This nested Ops binds that spec to FreeRTOS/PsychicMqttClient. The // Coordinator is owned and driven ONLY by the loop task (Core 1) from - // begin()/end(); the MQTT task (Core 0) communicates solely through the + // begin()/requestStop()/end()/loop(); the MQTT task (Core 0) communicates solely through the // _stop_requested/_stop_acked flags above. Methods are defined in the .cpp. class LifecycleOps : public MQTTLifecycle::Ops { public: @@ -587,6 +588,10 @@ public: void begin() override; void end() override; + // Core 1: request cooperative teardown without waiting or freeing resources. + // Call loop() to reap the acknowledgment; end() remains the OTA stop barrier. + void requestStop(); + bool isStopping() const { return _lifecycle.isStopInProgress(); } bool isRunning() const override { return _initialized; } void loop() override; void onPacketReceived(mesh::Packet *packet) override; @@ -681,8 +686,8 @@ public: int getConnectedBrokers() const; int getQueueSize() const; bool isReady() const; - /** True only after a CLEAN cooperative stop -- end() received the MQTT task's - * acknowledgment within the timeout. A timed-out/forced stop returns false so + /** True only after a CLEAN cooperative stop -- the MQTT task acknowledged + * its teardown before resources were released. A timed-out/forced stop returns false so * OTA flashing is withheld until a clean start/stop cycle. Mirrors * MQTTLifecycle::mayBeginFlash(); read on the loop task (Core 1). */ bool canFlashAfterStop() const { return _lifecycle.mayBeginFlash(); } diff --git a/test/fixtures/rs232_mqtt_runtime.cpp b/test/fixtures/rs232_mqtt_runtime.cpp index 0dddad4c9..0675cc210 100644 --- a/test/fixtures/rs232_mqtt_runtime.cpp +++ b/test/fixtures/rs232_mqtt_runtime.cpp @@ -112,6 +112,7 @@ struct MQTTNodeInfo { }; class MQTTBridge : public AbstractBridge { bool running = false; + bool stopping = false; public: inline static unsigned live = 0, fail_starts = 0; unsigned begins = 0, ends = 0, loops = 0, sent = 0, received = 0; @@ -127,9 +128,11 @@ public: if (fail_starts) { --fail_starts; return; } running = true; } - void end() override { ++ends; running = false; } + void end() override { ++ends; running = stopping = false; } + void requestStop() { stopping = running; } + bool isStopping() const { return stopping; } bool isRunning() const override { return running; } - void loop() override { ++loops; } + void loop() override { ++loops; if (stopping) { ++ends; running = stopping = false; } } void sendPacket(mesh::Packet* p) override { if (allow(p)) ++sent; } void onPacketReceived(mesh::Packet* p) { if (allow(p)) ++received; } }; @@ -171,6 +174,8 @@ struct Callbacks { virtual bool restartEspNowBridge() = 0; virtual bool isMqttBridgeRunning() = 0; virtual bool setMqttBridgeState(bool) = 0; + virtual bool requestMqttBridgeStop() = 0; + virtual bool isMqttBridgeStopping() = 0; virtual bool restartMqttBridge() = 0; }; class MyMesh : public Callbacks { @@ -277,7 +282,7 @@ int main() { assert(mesh.bridge->sent == 1 && mesh.bridge->loops == 1); assert(mesh.espnow_bridge.sent == 1 && mesh.espnow_bridge.loops == 1); assert(mesh.mqtt_bridge->received == 1 && mesh.mqtt_bridge->sent == 1); - assert(mesh.mqtt_bridge->loops == 0); // Its actual task services MQTT independently. + assert(mesh.mqtt_bridge->loops == 1); // Core 1 services stop acknowledgement only. assert(mesh.filter_checks == 4); packet.allowed = false; mesh.routeRx(&packet); mesh.routeTx(&packet); @@ -342,12 +347,34 @@ int main() { cli.set("bridge.enabled off", reply); assert(strcmp(reply, "OK") == 0 && !mesh.isEspNowBridgeRunning()); assert(mesh.isMqttBridgeRunning() && mesh.isRs232BridgeRunning()); + // A failed intent save must not signal an uncancellable worker stop. + cli.fail_save = true; cli.set("mqtt.enabled off", reply); - assert(strcmp(reply, "OK") == 0 && !mesh.isMqttBridgeRunning()); + assert(strstr(reply, "not saved; unchanged") && mesh._prefs.bridge_enabled); + assert(mesh.isMqttBridgeRunning() && !mesh.isMqttBridgeStopping()); + cli.fail_save = false; + cli.set("mqtt.enabled off", reply); + assert(strcmp(reply, "OK") == 0 && mesh.isMqttBridgeRunning()); + assert(mesh.isMqttBridgeStopping() && !mesh._prefs.bridge_enabled); + expect_get(cli, "mqtt.running", "> on"); + expect_get(cli, "mqtt.stopping", "> on"); + const unsigned stopping_saves = cli.saves; + cli.set("mqtt.enabled on", reply); + assert(strstr(reply, "MQTT is stopping") && !mesh._prefs.bridge_enabled); + assert(cli.saves == stopping_saves && !mesh.setMqttBridgeState(true)); + cli.set("mqtt.enabled off", reply); // a repeated request is accepted + assert(strcmp(reply, "OK") == 0 && mesh.isMqttBridgeStopping()); assert(mesh.isRs232BridgeRunning() && mesh.bridgesPreventSleep()); const unsigned uart_tx_before = mesh.bridge->sent; mesh.routeRx(&packet); mesh.serviceBridges(); assert(mesh.bridge->sent == uart_tx_before + 1); + assert(!mesh.isMqttBridgeRunning() && !mesh.isMqttBridgeStopping()); + expect_get(cli, "mqtt.stopping", "> off"); + cli.fail_save = true; + cli.set("mqtt.enabled on", reply); + assert(strstr(reply, "not saved; unchanged") && !mesh._prefs.bridge_enabled); + assert(!mesh.isMqttBridgeRunning()); + cli.fail_save = false; cli.set("espnow.enabled on", reply); assert(strcmp(reply, "OK") == 0); cli.set("mqtt.enabled on", reply); assert(strcmp(reply, "OK") == 0); mesh.serviceBridges(); assert(mesh.isEspNowBridgeRunning()); @@ -379,6 +406,8 @@ int main() { assert(mesh.isRs232BridgeRunning() && RS232Bridge::live == 1); } cli.set("rs232.enabled off", reply); cli.set("espnow.enabled off", reply); cli.set("mqtt.enabled off", reply); + assert(mesh.isMqttBridgeStopping() && mesh.bridgesPreventSleep()); + mesh.serviceBridges(); assert(!mesh.bridgesPreventSleep() && RS232Bridge::live == 0); cli.run("start ota", reply); cli.run("stop ota", reply); mesh.serviceBridges(); assert(!mesh.isRs232BridgeRunning() && !mesh.isMqttBridgeRunning() && !mesh.isEspNowBridgeRunning()); diff --git a/test/test_cli_settings_contract.py b/test/test_cli_settings_contract.py index 17759ccb9..9bd20c210 100644 --- a/test/test_cli_settings_contract.py +++ b/test/test_cli_settings_contract.py @@ -68,6 +68,7 @@ OBSERVER_QUERY_EXCEPTIONS = { 'mqtt.ntp.diag': 'NTP diagnostics', 'mqtt.presets': 'List of available presets', 'mqtt.running': 'Live state; set mqtt.enabled controls intent', + 'mqtt.stopping': 'Pending worker-owned shutdown; set mqtt.enabled off requests it', 'mqtt.stats': 'Runtime MQTT statistics', 'wifi.status': 'Live connection status', } diff --git a/test/test_common_cli_logging_transaction.py b/test/test_common_cli_logging_transaction.py index fb1d64bcd..e69937966 100644 --- a/test/test_common_cli_logging_transaction.py +++ b/test/test_common_cli_logging_transaction.py @@ -22,13 +22,15 @@ namespace mesh { void setUsbLoggingEnabled(bool value) { live_usb = value; ++usb_calls; } } struct Callbacks { - bool live_wifi = false, apply_ok = true; + bool live_wifi = false, apply_ok = true, stopping = false; unsigned calls = 0; bool setMqttBridgeState(bool value) { ++calls; if (apply_ok) live_wifi = value; return apply_ok; } + bool requestMqttBridgeStop() { return setMqttBridgeState(false); } + bool isMqttBridgeStopping() { return stopping; } }; struct CLI { Prefs prefs, persisted; @@ -88,6 +90,15 @@ int main() { assert(cli.persisted.usb_logging_enabled && cli.persisted.bridge_enabled); assert(live_usb && !cli.callbacks.live_wifi); assert(strstr(reply, "saved, but MQTT runtime change failed")); + // A restart during cleanup must not save ON or change either live output. + cli.callbacks.stopping = true; + cli.prefs = cli.persisted = {0, 0}; + live_usb = false; + const unsigned before_saves = cli.saves, before_calls = cli.callbacks.calls; + cli.set("logging.output both", reply); + assert(strstr(reply, "MQTT is stopping")); + assert(cli.saves == before_saves && cli.callbacks.calls == before_calls); + assert(!cli.prefs.bridge_enabled && !cli.persisted.bridge_enabled && !live_usb); #endif } ''' diff --git a/test/test_mqtt_transport_results.py b/test/test_mqtt_transport_results.py index 90d252fb5..000eb538b 100644 --- a/test/test_mqtt_transport_results.py +++ b/test/test_mqtt_transport_results.py @@ -26,6 +26,7 @@ def method(path, signature): PREAMBLE = r''' +#include #include #include #include @@ -90,6 +91,7 @@ struct MQTTBridge { }; MQTTSlot _slots[RUNTIME_MQTT_SLOTS]; volatile bool _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {}; + std::atomic _stop_requested{false}; esp_err_t reconnectSlotClient(int index); }; ''' @@ -152,12 +154,391 @@ int main() { sdk_callback=nullptr; reconnect_result=ESP_OK; assert(b.reconnectSlotClient(0)==ESP_OK && b._slot_attempt_pending[0]); + const int stopped_starts = start_calls, stopped_reconnects = reconnect_calls; + b._stop_requested = true; + assert(b.reconnectSlotClient(0)==ESP_ERR_INVALID_STATE); + assert(start_calls==stopped_starts && reconnect_calls==stopped_reconnects); return 0; } ''' +TASK_PREAMBLE = r''' +#include +#include +#include +#include +#include +#include +#include +#include +#define PORTABLE_MQTT_OBSERVER 1 +#define MQTT_DEBUG_PRINTLN(...) ((void)0) +using TickType_t = uint32_t; +constexpr int RUNTIME_MQTT_SLOTS = 3, WL_CONNECTED = 1; +constexpr uint32_t STATUS_RETRY_INTERVAL = 30000; +static uint32_t tick_ms = 1, first_tick = 0, first_millis = 0; +static uint64_t elapsed_ms = 0, stop_at_ms = UINT64_MAX; +static int scenario = 0; +static std::vector task_sleeps; +struct MQTTBridge; +static MQTTBridge* active_bridge = nullptr; +#define pdMS_TO_TICKS(ms) static_cast((ms) / tick_ms) +TickType_t xTaskGetTickCount() { + return first_tick + static_cast(elapsed_ms / tick_ms); +} +uint32_t millis() { return first_millis + static_cast(elapsed_ms); } +void vTaskDelay(TickType_t ticks); +struct { int status() { return WL_CONNECTED; } } WiFi; +bool mqttNtpRefreshDue(unsigned long, unsigned long, unsigned long) { return false; } +struct MQTTPrefs { + char mqtt_slot_preset[RUNTIME_MQTT_SLOTS][16] = {}; + bool mqtt_status_enabled = true; +}; +struct MQTTBridge { + struct Slot { + bool enabled = true, initial_connect_done = false, connected = false; + unsigned long last_reconnect_attempt = 0; + } _slots[RUNTIME_MQTT_SLOTS]; + struct Latch { + void noteGotIp() {} + bool consumeIfConnected(bool) { return false; } + } _ntp_reconnect_latch; + MQTTPrefs prefs; + MQTTPrefs* _obs = &prefs; + std::atomic _ntp_synced{true}; + bool _ntp_sync_pending = false, _slots_setup_done = false; + bool _ntp_force_requested = false, _ntp_force_result = false, _ntp_force_done = false; + bool _ntp_refresh_pending = false, _cached_has_connected_slots = true; + std::atomic _stop_requested{false}, _stop_acked{false}; + bool _slot_reconfigure_pending[RUNTIME_MQTT_SLOTS] = {true, true, true}; + bool _status_publish_pending[RUNTIME_MQTT_SLOTS] = {true, true, true}; + unsigned long _last_ntp_sync = 0, _ntp_refresh_retry_at = 0; + unsigned long _last_status_retry = 0, _last_status_publish = 0, _status_interval = 300000; + int _max_active_slots = RUNTIME_MQTT_SLOTS, _queue_count = 0; + int late_work = 0, reconfigures = 0, publishes = 0, queue_calls = 0; + std::vector setups; + std::vector setup_times; + std::vector teardown; + void work() { + if (_stop_requested) ++late_work; + assert(!_stop_acked); + } + void initializeWiFiInTask() {} + bool handleWiFiConnection(unsigned long) { work(); return false; } + bool syncTimeWithNTP(bool = false, bool = false) { + work(); + if (scenario == 6 || scenario == 7) { + _ntp_synced = true; + _stop_requested = true; + } + return true; + } + bool canActivateSlot(int) { work(); return true; } + bool isSlotReady(int, char*, size_t) { work(); return true; } + bool setupSlot(int index) { + work(); setups.push_back(index); setup_times.push_back(elapsed_ms); + if (scenario == 5) _stop_requested = true; + return true; + } + void applySlotPreset(int, const char*) { + work(); ++reconfigures; + if (scenario == 8) _stop_requested = true; + } + void publishStatusToSlot(int) { + work(); ++publishes; + if (scenario == 9) _stop_requested = true; + } + void maintainSlotConnections() { + work(); + if (scenario == 10) _stop_requested = true; + } + void processPacketQueue() { + work(); ++queue_calls; + if (scenario == 11) _stop_requested = true; + } + void checkConfigurationMismatch() { work(); } + void pollNtpRefresh(unsigned long) { work(); } + void refreshNTP() { work(); } + bool publishStatus() { work(); ++publishes; return true; } + void updateCachedConnectionStatus() { work(); } + void cancelNtpRefresh() { + assert(_stop_requested && !_stop_acked); + teardown.push_back("ntp"); + } + void teardownSlot(int index) { + assert(_stop_requested && !_stop_acked); + teardown.push_back("slot" + std::to_string(index)); + } + void destroySlotClients() { + assert(!_stop_acked); + assert(teardown == std::vector({"ntp", "slot0", "slot1", "slot2"})); + teardown.push_back("destroy"); + } + bool waitUnlessStopping(uint32_t delay_ms); + void mqttTaskLoop(); +}; +void vTaskDelay(TickType_t ticks) { + assert(ticks > 0); + const uint32_t duration = ticks * tick_ms; + task_sleeps.push_back(duration); + elapsed_ms += duration; + if (elapsed_ms >= stop_at_ms) active_bridge->_stop_requested = true; + // The no-stop scenario exits after one complete worker iteration. Stop is + // raised during its ordinary idle delay, after all setup and publications. + if (scenario == 3 && active_bridge->setups.size() == RUNTIME_MQTT_SLOTS) + active_bridge->_stop_requested = true; + assert(elapsed_ms < 30000); // bound a missing cooperative exit deterministically +} +''' + +TASK_MAIN = r''' +int main(int argc, char** argv) { + assert(argc == 4); + scenario = std::atoi(argv[1]); + tick_ms = static_cast(std::atoi(argv[2])); + if (std::atoi(argv[3])) { + first_tick = UINT32_MAX - 511; + first_millis = UINT32_MAX - 511; + } + MQTTBridge bridge; + active_bridge = &bridge; + switch (scenario) { + case 0: stop_at_ms = 187; break; // stop while WiFi settles + case 1: stop_at_ms = 1173; break; // stop during first TLS stagger + case 2: stop_at_ms = 6191; break; // stop during second TLS stagger + case 3: break; // preserve every normal delay + case 4: bridge._stop_requested = true; break; + case 5: break; // setup callback requests stop + case 6: bridge._ntp_force_requested = true; break; + case 7: bridge._ntp_sync_pending = true; bridge._ntp_synced = false; break; + case 8: case 9: case 10: case 11: break; // stop raised during worker callbacks + default: assert(false); + } + bridge.mqttTaskLoop(); + assert(bridge._stop_acked && bridge.late_work == 0); + assert(bridge.teardown == std::vector({ + "ntp", "slot0", "slot1", "slot2", "destroy"})); + if (scenario == 3) { + assert(bridge.setups == std::vector({0, 1, 2})); + assert(bridge.setup_times == std::vector({1000, 6000, 11000})); + assert(bridge.reconfigures == RUNTIME_MQTT_SLOTS); + assert(bridge.publishes >= RUNTIME_MQTT_SLOTS && bridge.queue_calls == 1); + } else { + const size_t wanted = scenario >= 8 ? RUNTIME_MQTT_SLOTS + : scenario == 0 || scenario == 4 || scenario >= 6 ? 0 + : scenario == 2 ? 2 : 1; + assert(bridge.setups.size() == wanted); + if (scenario >= 8) { + assert(bridge.reconfigures == (scenario == 8 ? 1 : RUNTIME_MQTT_SLOTS)); + assert(bridge.publishes == (scenario == 8 ? 0 : scenario == 9 ? 1 : RUNTIME_MQTT_SLOTS)); + assert(bridge.queue_calls == (scenario == 11 ? 1 : 0)); + } else { + assert(bridge.reconfigures == 0 && bridge.publishes == 0 && bridge.queue_calls == 0); + } + const uint32_t poll_bound = std::max(uint32_t{20}, tick_ms); + assert(std::all_of(task_sleeps.begin(), task_sleeps.end(), + [poll_bound](uint32_t duration) { return duration <= poll_bound; })); + if (scenario == 4) assert(elapsed_ms == 0 && task_sleeps.empty()); + else if (scenario >= 8) assert(elapsed_ms == 11000); + else if (scenario >= 5) assert(elapsed_ms == 1000); + else assert(elapsed_ms >= stop_at_ms && elapsed_ms - stop_at_ms < poll_bound); + } + // A completed zero-duration wait preserves its caller's stop decision. + const auto sleeps_before = task_sleeps.size(); + bridge._stop_requested = false; + assert(bridge.waitUnlessStopping(0)); + bridge._stop_requested = true; + assert(!bridge.waitUnlessStopping(0)); + assert(task_sleeps.size() == sleeps_before); +} +''' + +ASYNC_STOP_PREAMBLE = r''' +#include +#include +#include +#include +#include "helpers/MQTTLifecycle.h" +#define ESP_PLATFORM 1 +#define MQTT_DEBUG_PRINTLN(...) ((void)0) +#define pdMS_TO_TICKS(ms) (ms) +constexpr int RUNTIME_MQTT_SLOTS = 3; +static constexpr uint32_t MQTT_STOP_TIMEOUT_BASE_MS = 5000; +static constexpr uint32_t MQTT_STOP_TIMEOUT_PER_SLOT_MS = 8000; +static uint32_t clock_ms = 0, sleeps = 0; +static uint64_t elapsed_ms = 0, ack_after_ms = UINT64_MAX; +uint32_t millis() { return clock_ms; } +struct MQTTBridge; +static MQTTBridge* s_mqtt_bridge_instance = nullptr; +static MQTTBridge* bridge_waiting = nullptr; +void vTaskDelay(uint32_t duration); +struct MQTTBridge { + struct Slot { bool enabled = true; } _slots[RUNTIME_MQTT_SLOTS]; + struct LifecycleOps : MQTTLifecycle::Ops { + MQTTBridge* _b; + explicit LifecycleOps(MQTTBridge* bridge) : _b(bridge) {} + uint32_t nowMs() override; + void startTask() override; + void deliverStop() override; + void releaseResources() override; + void onStopComplete(bool clean) override; + }; + struct RecordingOps : LifecycleOps { + int signals = 0; + explicit RecordingOps(MQTTBridge* bridge) : LifecycleOps(bridge) {} + void deliverStop() override { ++signals; LifecycleOps::deliverStop(); } + } ops{this}; + MQTTLifecycle::Coordinator _lifecycle{ops, 1}; + std::atomic _stop_requested{false}, _stop_acked{false}; + bool _initialized = true, resources_owned = true; + bool _slots_setup_done = true, _staged_raw_valid = true; + bool _ntp_estimate_requested = true, _ntp_estimate_done = true, _ntp_estimate_ok = true; + uint32_t _ntp_estimate_epoch = 123; + int releases = 0, cancels = 0, completions = 0; + bool clean_completion = false; + MQTTBridge() { + assert(_lifecycle.requestStart()); + assert(_lifecycle.onTaskStarted()); + s_mqtt_bridge_instance = this; + } + void cancelNtpRefresh() { ++cancels; } + void requestStop(); + void end(); + void finishStopped(); + void loop(); +}; +void MQTTBridge::LifecycleOps::releaseResources() { + assert(_b->_stop_acked.load() || _b->_lifecycle.stopTimedOut()); + assert(_b->resources_owned && _b->_initialized); + _b->resources_owned = false; + ++_b->releases; +} +void MQTTBridge::LifecycleOps::onStopComplete(bool clean) { + ++_b->completions; + _b->clean_completion = clean; +} +void vTaskDelay(uint32_t duration) { + assert(duration > 0); + ++sleeps; + clock_ms += duration; + elapsed_ms += duration; + if (elapsed_ms >= ack_after_ms) + bridge_waiting->_stop_acked.store(true, std::memory_order_release); + assert(elapsed_ms < 60000); +} +static void assert_finished(MQTTBridge& bridge, bool clean) { + assert(!bridge._initialized && !bridge.resources_owned); + assert(bridge.releases == 1 && bridge.completions == 1); + assert(bridge.clean_completion == clean); + assert(!bridge._lifecycle.isStopInProgress() && bridge._lifecycle.mayRestart()); + assert(bridge._lifecycle.mayBeginFlash() == clean); + assert(!bridge._slots_setup_done && !bridge._staged_raw_valid); + assert(!bridge._ntp_estimate_requested && !bridge._ntp_estimate_done); + assert(!bridge._ntp_estimate_ok && bridge._ntp_estimate_epoch == 0); + assert(bridge.cancels == 1); +} +''' + +ASYNC_STOP_MAIN = r''' +int main() { + { + MQTTBridge bridge; + bridge._stop_acked = true; // stale prior-epoch ACK must be cleared exactly once + bridge.requestStop(); + assert(sleeps == 0 && elapsed_ms == 0 && bridge.ops.signals == 1); + assert(bridge._stop_requested && !bridge._stop_acked); + assert(bridge._initialized && bridge.resources_owned && bridge.releases == 0); + assert(s_mqtt_bridge_instance == nullptr); + const auto timeout = bridge._lifecycle.stopTimeoutMs(); + assert(timeout > 5000); + for (int pass = 0; pass < 10; ++pass) { + clock_ms += timeout; // ordinary loop must never force-free a slow worker + bridge.requestStop(); + bridge.loop(); + assert(bridge.ops.signals == 1 && sleeps == 0); + assert(bridge._initialized && bridge.resources_owned && bridge.releases == 0); + assert(bridge._lifecycle.isStopInProgress() && !bridge._lifecycle.stopTimedOut()); + } + bridge._stop_acked.store(true, std::memory_order_release); + bridge.requestStop(); + assert(bridge._stop_acked && bridge.ops.signals == 1); // duplicate cannot revoke ACK + bridge.loop(); + assert_finished(bridge, true); + bridge.loop(); + bridge.requestStop(); + bridge.end(); + assert(bridge.releases == 1 && bridge.ops.signals == 1 && bridge.cancels == 1); + } + for (int at_deadline = 0; at_deadline < 2; ++at_deadline) { + MQTTBridge bridge; + clock_ms = UINT32_MAX - 500; elapsed_ms = 0; sleeps = 0; + bridge_waiting = &bridge; + bridge.requestStop(); + ack_after_ms = at_deadline ? bridge._lifecycle.stopTimeoutMs() : 8000; + bridge.end(); // synchronous OTA barrier joins the existing request + assert(elapsed_ms == ack_after_ms && sleeps > 0 && bridge.ops.signals == 1); + assert(!bridge._lifecycle.stopTimedOut()); + assert_finished(bridge, true); + } + { + MQTTBridge bridge; + clock_ms = 0; elapsed_ms = 0; sleeps = 0; + bridge_waiting = &bridge; ack_after_ms = UINT64_MAX; + bridge.end(); // reviewed timeout remains confined to the explicit barrier + assert(bridge.ops.signals == 1 && bridge._lifecycle.stopTimedOut()); + assert(elapsed_ms == bridge._lifecycle.stopTimeoutMs()); + assert_finished(bridge, false); + } +} +''' + + class MqttTransportResultsTests(unittest.TestCase): + def test_actual_async_stop_preserves_resources_until_ack_and_ota_joins(self): + compiler = shutil.which("g++") or shutil.which("c++") + self.assertIsNotNone(compiler, "a host C++ compiler is required") + bridge = "src/helpers/bridges/MQTTBridge.cpp" + source = ASYNC_STOP_PREAMBLE + "\n".join(method(bridge, signature) for signature in ( + "static inline uint32_t mqttStopTimeoutForSlots(int slots)", + "void MQTTBridge::requestStop()", "void MQTTBridge::end()", + "void MQTTBridge::finishStopped()", "void MQTTBridge::loop()", + "uint32_t MQTTBridge::LifecycleOps::nowMs()", + "void MQTTBridge::LifecycleOps::startTask()", + "void MQTTBridge::LifecycleOps::deliverStop()")) + ASYNC_STOP_MAIN + with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-async-stop-") as temp: + path = Path(temp) / "stop.cpp" + path.write_text(source, encoding="utf-8") + exe = Path(temp) / ("stop.exe" if os.name == "nt" else "stop") + result = subprocess.run([compiler, "-std=c++17", "-Wall", "-Wextra", + "-I", str(ROOT / "src"), str(path), "-o", str(exe)], + text=True, capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr) + result = subprocess.run([str(exe)], text=True, capture_output=True, timeout=5) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_actual_worker_interrupts_startup_and_stagger_without_late_work(self): + compiler = shutil.which("g++") or shutil.which("c++") + self.assertIsNotNone(compiler, "a host C++ compiler is required") + bridge = "src/helpers/bridges/MQTTBridge.cpp" + source = TASK_PREAMBLE + method(bridge, + "bool MQTTBridge::waitUnlessStopping(uint32_t delay_ms)") + source += method(bridge, "void MQTTBridge::mqttTaskLoop()") + TASK_MAIN + with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-stop-") as temp: + path = Path(temp) / "worker.cpp" + path.write_text(source, encoding="utf-8") + exe = Path(temp) / ("worker.exe" if os.name == "nt" else "worker") + result = subprocess.run([compiler, "-std=c++17", "-Wall", "-Wextra", + str(path), "-o", str(exe)], text=True, capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr) + for scenario in range(12): + for tick_ms in (1, 10, 50): + for wrap in (0, 1): + with self.subTest(scenario=scenario, tick_ms=tick_ms, wrap=wrap): + result = subprocess.run([str(exe), str(scenario), str(tick_ms), str(wrap)], + text=True, capture_output=True, timeout=5) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + def test_real_adapter_and_bridge_methods_for_both_idf_layouts(self): compiler = shutil.which("g++") or shutil.which("c++") self.assertIsNotNone(compiler, "a host C++ compiler is required") diff --git a/test/test_rs232_mqtt_runtime.py b/test/test_rs232_mqtt_runtime.py index 53c95d412..9eef0e6c0 100644 --- a/test/test_rs232_mqtt_runtime.py +++ b/test/test_rs232_mqtt_runtime.py @@ -35,7 +35,8 @@ class RS232MQTTRuntimeTests(unittest.TestCase): "bool setRs232BridgeState(bool enable)", "bool restartRs232Bridge()", "bool startSharedEspNowBridgeIfReady()", "bool isEspNowBridgeRunning()", "bool setEspNowBridgeState(bool enable)", "bool restartEspNowBridge()", - "bool isMqttBridgeRunning()", "bool setMqttBridgeState(bool enable)", + "bool isMqttBridgeRunning()", "bool requestMqttBridgeStop()", "bool isMqttBridgeStopping()", + "bool setMqttBridgeState(bool enable)", "bool restartMqttBridge()", "bool setBridgeState(bool enable)", "bool restartBridge()", )) defaults_start = implementation.index("// bridge defaults") @@ -49,7 +50,7 @@ class RS232MQTTRuntimeTests(unittest.TestCase): start = implementation.index("#ifdef WITH_MQTT_BRIDGE\n", start) end = implementation.index(" if (_logging)", start) routing[name] = implementation[start:end] - service_start = implementation.index("#if defined(WITH_ESPNOW_BRIDGE)", + service_start = implementation.index("#if defined(WITH_MQTT_BRIDGE)\n // MQTT owns TLS", implementation.index("MyMesh::servicePostMeshLoop()")) service_end = implementation.index(" if (next_flood_advert", service_start) pending_start = implementation.index("#if defined(WITH_BRIDGE)", @@ -80,7 +81,8 @@ class RS232MQTTRuntimeTests(unittest.TestCase): 'if (configKeyEquals(config, "bridge.uart"))', )) gets += "\nelse " + "\nelse ".join(extract_braced(observer, signature) for signature in ( - 'if (strcmp(config, "mqtt.enabled") == 0)', 'if (strcmp(config, "mqtt.running") == 0)')) + 'if (strcmp(config, "mqtt.enabled") == 0)', 'if (strcmp(config, "mqtt.running") == 0)', + 'if (strcmp(config, "mqtt.stopping") == 0)')) ota_start = cli.index(' } else if (memcmp(command, "start ota", 9)') ota_end = cli.index(' } else if (memcmp(command, "clock", 5)', ota_start) code = (ROOT / "test/fixtures/rs232_mqtt_runtime.cpp").read_text(encoding="ascii") @@ -128,7 +130,8 @@ class RS232MQTTRuntimeTests(unittest.TestCase): "RS232Bridge* createRS232Bridge()", "bool beginRS232Bridge()", "bool endRS232Bridge()", "bool rs232BridgeEnabled() const", "bool isBridgeRunning() const override", "bool isRs232BridgeRunning()", "bool setRs232BridgeState(bool enable)", - "bool restartRs232Bridge()", "bool isMqttBridgeRunning()", "bool setBridgeState(bool enable)", + "bool restartRs232Bridge()", "bool isMqttBridgeRunning()", "bool requestMqttBridgeStop()", + "bool isMqttBridgeStopping()", "bool setBridgeState(bool enable)", "bool restartBridge()")) code = (ROOT / "test/fixtures/rs232_mqtt_runtime.cpp").read_text(encoding="ascii") code = code[:code.index("int main() {")] @@ -144,7 +147,7 @@ class RS232MQTTRuntimeTests(unittest.TestCase): boot_start = implementation.index('mesh::hilStartupTrace("mesh_bridge_begin");') boot_start = implementation.index("#if defined(WITH_BRIDGE)", boot_start) boot_end = implementation.index('mesh::hilStartupTrace("mesh_bridge_ready");', boot_start) - service_start = implementation.index("#if defined(WITH_ESPNOW_BRIDGE)", implementation.index("MyMesh::servicePostMeshLoop()")) + service_start = implementation.index("#if defined(WITH_MQTT_BRIDGE)\n // MQTT owns TLS", implementation.index("MyMesh::servicePostMeshLoop()")) service_end = implementation.index(" if (next_flood_advert", service_start) pending_start = implementation.index("#if defined(WITH_BRIDGE)", implementation.index("bool MyMesh::hasPendingWork()")) pending_end = implementation.index(" if (radio_driver.isWatchdogObserving()", pending_start) @@ -199,6 +202,8 @@ class RS232MQTTRuntimeTests(unittest.TestCase): cli.run("stop ota", reply); assert(!mesh._cli.board.ota && mesh.isMqttBridgeRunning() && Serial2.begins == uart_starts); cli.set("rs232.enabled off", reply); cli.set("mqtt.enabled off", reply); + assert(!mesh.bridge && mesh.isMqttBridgeStopping() && mesh.bridgesPreventSleep()); + mesh.serviceBridges(); assert(!mesh.bridge && !mesh.isMqttBridgeRunning() && !mesh.bridgesPreventSleep()); cli.set("rs232.enabled on", reply); mesh.mqtt_bridge->fail_starts = 1; cli.set("mqtt.enabled on", reply);