diff --git a/MQTT_IMPLEMENTATION.md b/MQTT_IMPLEMENTATION.md index 56f43569a..b46714877 100644 --- a/MQTT_IMPLEMENTATION.md +++ b/MQTT_IMPLEMENTATION.md @@ -883,7 +883,7 @@ the radio actually performs in that case. ### NTP Time Synchronization - Automatic time synchronization with NTP servers (required for JWT authentication) - Default primary: `pool.ntp.org`; built-in fallbacks (tried sequentially on failure): `time.google.com`, `time.cloudflare.com`, `time.aws.com`, `time.nist.gov` -- Periodic time updates (every hour) on the effective primary only; system time is kept in UTC +- Periodic time updates (every hour) through the same validated probe, one attempt per server in list order; system time is kept in UTC - Replies are validated before they are trusted: the datagram must be a full-length NTPv3/v4 server reply from the queried address and port, from a synchronised server (stratum 1-15, no leap alarm), echoing the random transmit timestamp of the request, with a plausible epoch. Anything else is discarded and the clock, the RTC and JWT issuance are left alone - Configure and diagnose with `set mqtt.ntp` / `get mqtt.ntp` / `get mqtt.ntp.diag` — see [MQTT Shared Commands](#mqtt-shared-commands) diff --git a/examples/simple_repeater/MyMesh.cpp b/examples/simple_repeater/MyMesh.cpp index 75d9fb4ee..93dbe2790 100644 --- a/examples/simple_repeater/MyMesh.cpp +++ b/examples/simple_repeater/MyMesh.cpp @@ -1699,6 +1699,15 @@ void MyMesh::loop() { // MQTT processing runs in a separate FreeRTOS task on Core 0, so we don't call bridge.loop() here mesh::Mesh::loop(); +#ifdef WITH_MQTT_BRIDGE + // A timed-out stop keeps the bridge down until the MQTT task acknowledges it; + // once that late ack lands, restart the bridge that was meant to be running. + if (_bridge_resume_pending && bridge && bridge->stopAcknowledgedLate()) { + Serial.println("MQTT: stop acknowledged late - resuming bridge"); + setBridgeState(true); + } +#endif + #ifdef WITH_BRIDGE // bridge.loop() is now handled by FreeRTOS task on Core 0 - no need to call it here #endif @@ -1752,8 +1761,9 @@ void MyMesh::loop() { // resume the bridge instead of flashing under uncertain ownership. if (bridge && !bridge->canFlashAfterStop()) { Serial.println("OTA: aborted, MQTT stop did not complete cleanly - resuming bridge"); - otaAlert("OTA aborted: MQTT stop unclean, bridge resumed"); setBridgeState(true); + otaAlert(bridge->isRunning() ? "OTA aborted: MQTT stop unclean, bridge resumed" + : "OTA aborted: MQTT stop unproven, bridge resumes when it completes"); } else if (!_cli.getBoard()->otaFromManifest(getFirmwareVer(), false, ota_reply)) { Serial.print("OTA: aborted, resuming bridge - "); Serial.println(ota_reply); char ota_alert_msg[160]; diff --git a/examples/simple_repeater/MyMesh.h b/examples/simple_repeater/MyMesh.h index 539a17154..3dd8587b6 100644 --- a/examples/simple_repeater/MyMesh.h +++ b/examples/simple_repeater/MyMesh.h @@ -137,6 +137,8 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks ESPNowBridge bridge; #elif defined(WITH_MQTT_BRIDGE) MQTTBridge* bridge; + // begin() was refused after an unproven stop; restart once the task acknowledges. + bool _bridge_resume_pending = false; #endif #ifdef WITH_SNMP MeshSNMPAgent _snmp_agent; @@ -368,6 +370,7 @@ public: bridge->begin(); #ifdef WITH_MQTT_BRIDGE _alerter.setBridge(bridge); + _bridge_resume_pending = !bridge->isRunning() && bridge->isStopUnproven(); #endif } else @@ -375,6 +378,7 @@ public: bridge->end(); #ifdef WITH_MQTT_BRIDGE _alerter.setBridge(nullptr); + _bridge_resume_pending = false; #endif } } @@ -400,6 +404,9 @@ public: bridge->setStatsSources(this, _radio, _cli.getBoard(), _ms); #endif bridge->begin(); +#ifdef WITH_MQTT_BRIDGE + _bridge_resume_pending = !bridge->isRunning() && bridge->isStopUnproven(); +#endif } void restartBridgeSlot(int slot) override { diff --git a/examples/simple_room_server/MyMesh.cpp b/examples/simple_room_server/MyMesh.cpp index e6bed7d60..956b59596 100644 --- a/examples/simple_room_server/MyMesh.cpp +++ b/examples/simple_room_server/MyMesh.cpp @@ -1514,6 +1514,15 @@ void MyMesh::loop() { // Check radio FIRST to ensure we don't miss incoming packets // MQTT processing can take time, so we prioritize radio reception mesh::Mesh::loop(); + +#ifdef WITH_MQTT_BRIDGE + // A timed-out stop keeps the bridge down until the MQTT task acknowledges it; + // once that late ack lands, restart the bridge that was meant to be running. + if (_bridge_resume_pending && bridge && bridge->stopAcknowledgedLate()) { + Serial.println("MQTT: stop acknowledged late - resuming bridge"); + setBridgeState(true); + } +#endif #ifdef WITH_MQTT_BRIDGE // bridge.loop() is now handled by FreeRTOS task on Core 0 - no need to call it here #endif diff --git a/examples/simple_room_server/MyMesh.h b/examples/simple_room_server/MyMesh.h index 47f2f68d7..0139fa5a7 100644 --- a/examples/simple_room_server/MyMesh.h +++ b/examples/simple_room_server/MyMesh.h @@ -205,6 +205,8 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks #endif #ifdef WITH_MQTT_BRIDGE MQTTBridge* bridge; + // begin() was refused after an unproven stop; restart once the task acknowledges. + bool _bridge_resume_pending = false; #endif #ifdef WITH_SNMP MeshSNMPAgent _snmp_agent; @@ -385,6 +387,7 @@ public: bridge->begin(); #ifdef WITH_MQTT_BRIDGE _alerter.setBridge(bridge); + _bridge_resume_pending = !bridge->isRunning() && bridge->isStopUnproven(); #endif } else @@ -392,6 +395,7 @@ public: bridge->end(); #ifdef WITH_MQTT_BRIDGE _alerter.setBridge(nullptr); + _bridge_resume_pending = false; #endif } } @@ -416,6 +420,9 @@ public: bridge->setStatsSources(this, _radio, _cli.getBoard(), _ms); #endif bridge->begin(); +#ifdef WITH_MQTT_BRIDGE + _bridge_resume_pending = !bridge->isRunning() && bridge->isStopUnproven(); +#endif } void restartBridgeSlot(int slot) override { diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index d1c866e22..2f8d69ef1 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -464,6 +464,14 @@ void MQTTBridge::applyWifiPowerSave() { bool MQTTBridge::stopUnprovenLatched() { return s_stop_unproven; } +bool MQTTBridge::stopAcknowledgedLate() const { +#ifdef ESP_PLATFORM + return _lifecycle.isStopUnproven() && _stop_acked.load(std::memory_order_acquire); +#else + return false; +#endif +} + uint8_t MQTTBridge::getLastWifiDisconnectReason() { return s_wifi_disconnect_reason; } unsigned long MQTTBridge::getLastWifiDisconnectTime() { return s_wifi_disconnect_time; } @@ -1644,9 +1652,9 @@ void MQTTBridge::mqttTaskLoop() { // Periodic configuration check (throttled to avoid spam) checkConfigurationMismatch(); - // Periodic NTP refresh (every hour) — lightweight, non-blocking. - // Uses async SNTP instead of the heavy syncTimeWithNTP() which blocks Core 0 - // for up to 20+ seconds with DNS lookups, UDP sockets, and retry loops. + // Periodic NTP refresh (every hour): a validated probe, one attempt per + // server. Blocks this task for about one probe timeout per server that does + // not answer (see refreshNTP()). if (WiFi.status() == WL_CONNECTED && now - _last_ntp_sync > 3600000) { refreshNTP(); } @@ -4493,10 +4501,11 @@ void MQTTBridge::storeRawRadioData(const uint8_t* raw_data, int len, float snr, // let it set the system clock asynchronously — an acceptance path with NO // validation of its own (see the syncTimeWithNTP() note below), running every // hour for the life of the node. It now goes through the same validated probe -// as every other sync, which costs the MQTT task about a second per attempt and -// touches nothing until a reply passes every check. +// as every other sync and touches nothing until a reply passes every check. +// One attempt per server bounds the blocking walk on a network that drops +// UDP/123 to about one probe timeout per server, instead of three. void MQTTBridge::refreshNTP() { - syncTimeWithNTP(/*force=*/true, /*primary_only=*/false); + syncTimeWithNTP(/*force=*/true, /*primary_only=*/false, /*attempts_per_server=*/1); } // One validated NTP exchange with one named server, on a fresh ephemeral socket. @@ -4599,7 +4608,7 @@ bool MQTTBridge::probeNtpServer(const char* server, uint32_t min_epoch, return accepted; } -bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) { +bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only, int attempts_per_server) { if (!WiFi.isConnected()) { MQTT_DEBUG_PRINTLN("Cannot sync time - WiFi not connected"); return false; @@ -4634,7 +4643,7 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) { const uint32_t kMinValidEpoch = kNtpMinValidEpoch; const char* ntp_server_used = nullptr; - const int kMaxNtpRetriesPerServer = 2; + const int kMaxNtpRetriesPerServer = attempts_per_server > 0 ? attempts_per_server : 1; for (int s = 0; s < server_count && !ntp_ok; s++) { const char* server = servers[s]; diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 8d180a48b..e733044ff 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -743,6 +743,9 @@ public: // the bridge is down, nothing was released, and it will not restart until the // task acknowledges late (pollLateStopAck) or the node reboots. bool isStopUnproven() const { return _lifecycle.isStopUnproven(); } + // The unproven stop has since been acknowledged, so begin() will release the + // withheld resources and start. Loop task only. + bool stopAcknowledgedLate() const; // Survives end() clearing the diagnostic singleton, so `get mqtt.status` can // still explain why a stopped bridge will not come back without a reboot. static bool stopUnprovenLatched(); @@ -787,7 +790,7 @@ public: * mistyped hostname fails fast instead of blocking through the whole fallback list. * Performs blocking NTP I/O and must only be called from the MQTT task (Core 0). * Other tasks (e.g. the CLI on Core 1) must use requestForcedNtpSync() instead. */ - bool syncTimeWithNTP(bool force = false, bool primary_only = false); + bool syncTimeWithNTP(bool force = false, bool primary_only = false, int attempts_per_server = 2); /** Request a forced NTP sync from another task (e.g. CLI on Core 1). Marshals the * work onto the MQTT task so all NTP I/O stays on Core 0, then blocks up to * timeout_ms for the result. Returns true if the sync succeeded, false on failure,