fix(mqtt): resume after a late stop ack; bound the hourly NTP refresh
- A stop that timed out into StopUnproven refused begin(), and only a later begin() honoured the task's late acknowledgement. Nothing called it, so restartBridge() (any `set mqtt...` restart) or an aborted OTA left the observer offline for the rest of the boot. MyMesh now records that a start was refused and restarts the bridge once stopAcknowledgedLate() reports the ack. The OTA-abort alert no longer claims "bridge resumed" when the restart was refused. - refreshNTP() now runs the validated probe with one attempt per server. With two attempts and a 1 s pause it blocked the MQTT task ~18 s every hour on networks that drop UDP/123, where the old async SNTP cost nothing. Loop comment and docs updated to match.
This commit is contained in:
1 parent
0ff3408564
commit
3c03254aa2
7 files changed
+56
-11
No files matched your search
@@ -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)
|
||||
|
||||
|
||||
@@ -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];
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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];
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in new issue
Block a user