diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 29edc89e..c9c3f333 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -1,4 +1,6 @@ #include "MQTTBridge.h" +#include "helpers/MQTTSlotEvents.h" +#include "helpers/ObserverSlotStatsJson.h" #include "MQTTErrorLabels.h" #include "../WifiPowerSavePolicy.h" #include "../MQTTConnectionPolicy.h" @@ -34,10 +36,10 @@ #endif // Effective MQTT origin: empty mqtt_origin follows node_name; otherwise mqtt_origin override (quotes stripped). -static void applyEffectiveOrigin(const NodePrefs* np, const MQTTPrefs* obs, char* dest, size_t dest_size) { - if (!np || !obs || !dest || dest_size == 0) return; +static void applyEffectiveOrigin(const char* node_name, const MQTTPrefs* obs, char* dest, size_t dest_size) { + if (!node_name || !obs || !dest || dest_size == 0) return; if (obs->mqtt_origin[0] == '\0') { - strncpy(dest, np->node_name, dest_size - 1); + strncpy(dest, node_name, dest_size - 1); } else { strncpy(dest, obs->mqtt_origin, dest_size - 1); } @@ -94,8 +96,7 @@ const char* MQTTBridge::effectiveNtpPrimary(const MQTTPrefs* obs) { } void MQTTBridge::refreshOriginFromPrefs() { - if (!_prefs) return; - applyEffectiveOrigin(_prefs, _obs, _origin, sizeof(_origin)); + applyEffectiveOrigin(_radio_metadata.node_name, _obs, _origin, sizeof(_origin)); } void MQTTBridge::getEffectiveMqttOrigin(const NodePrefs* np, const MQTTPrefs* obs, char* buf, size_t buf_size) { @@ -104,7 +105,7 @@ void MQTTBridge::getEffectiveMqttOrigin(const NodePrefs* np, const MQTTPrefs* ob buf[0] = '\0'; return; } - applyEffectiveOrigin(np, obs, buf, buf_size); + applyEffectiveOrigin(np->node_name, obs, buf, buf_size); } static bool isNetworkConfigValid(const MQTTPrefs* obs) { @@ -209,10 +210,7 @@ void* MQTTBridge::JsonScratchAllocator::reallocate(void* ptr, size_t new_size) { return psram_realloc(ptr, new_size); } -// Latched by end() when a stop passes its deadline unacknowledged. end() clears -// the diagnostic singleton, so without this `get mqtt.status` could only say -// "not running" and an operator would have no way to learn that the bridge is -// down for the rest of the boot and why. +// Loop-task latch explaining why a stopped bridge cannot yet restart. static bool s_stop_unproven = false; #ifdef MQTT_MEMORY_DEBUG @@ -230,8 +228,59 @@ static void agentLogHeap(const char* location, const char* message, const char* // #endregion #endif -// Singleton for formatMqttStatusReply (set in begin(), cleared in end()) -static MQTTBridge* s_mqtt_bridge_instance = nullptr; +ObserverMailbox MQTTBridge::_published; + +void MQTTBridge::publishRuntimeSnapshot() { + RuntimeSnapshot snapshot; + snapshot.sampled_ms = millis(); + snapshot.sequence = ++_snapshot_sequence; + snapshot.desired_revision = _configuration.revision(); + snapshot.applied_revision = _applied_revision; + snapshot.running = !_stop_requested.load(std::memory_order_acquire); + snapshot.queue = getQueueSize(); + snapshot.skipped = _skipped_publishes; + snapshot.filtered = _filtered_packets.load(std::memory_order_relaxed); + snapshot.event_overflows = _slot_events.overflows(); + snapshot.free_heap = ESP.getFreeHeap(); + snapshot.max_heap = ESP.getMaxAllocHeap(); +#if defined(WITH_MQTT_NEIGHBORS) + snapshot.neighbors_phase = _neighbors_phase.load(std::memory_order_relaxed); + snapshot.neighbors_result = _neighbors_last_result.load(std::memory_order_relaxed); + snapshot.neighbors_seconds = _neighbors_secs_until_next.load(std::memory_order_relaxed); +#endif + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { + const MQTTSlot& slot = _slots[i]; + SlotRuntimeSnapshot& out = snapshot.slots[i]; + out.configured = slot.enabled || slot.preset; + out.enabled = slot.enabled; + out.client = slot.client != nullptr; + out.connected = slot.connected; + out.attempted = slot.initial_connect_done; + out.ready = isSlotReady(i); + out.circuit_breaker_tripped = slot.circuit_breaker_tripped; + out.client_state = slot.client_state; + snprintf(out.name, sizeof(out.name), "%s", slot.preset ? slot.preset->name : slot.enabled ? "custom" : "none"); + snprintf(out.state, sizeof(out.state), "%s", !slot.enabled ? "inactive" : !out.ready ? "wait" : + slot.connected ? "ok" : slot.circuit_breaker_tripped ? "fail" : "disc"); + out.filter_mask = _obs->mqtt_slot_packet_filter[i]; + out.disconnect_count = slot.disconnect_count; + out.first_disconnect_time = slot.first_disconnect_time; + out.current_outage_started_ms = slot.current_outage_started_ms; + out.last_error_time = slot.last_error_time; + out.last_tls_err = slot.last_tls_err; + out.last_tls_stack_err = slot.last_tls_stack_err; + out.last_sock_errno = slot.last_sock_errno; + out.last_connack_code = slot.last_connack_code; + if (slot.enabled && slot.connected) ++snapshot.connected; + if (slot.client && slot.client_state != ClientState::Quarantined) { + out.publish_ok = slot.client->getPublishOk(); + out.publish_err = slot.client->getPublishErr(); + snapshot.outbox += slot.client->getOutboxSize(); + snapshot.outbox_drops += slot.client->getOutboxDrops(); + } + } + _published.publish(snapshot); +} unsigned long MQTTBridge::getWifiConnectedAtMillis() { return activeNetworkLink().connectedAtMillis(); @@ -256,52 +305,25 @@ static void formatDuration(char* buf, size_t len, uint32_t secs) { void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPrefs* obs) { if (buf == nullptr || bufsize == 0) return; const char* msgs = (obs && obs->mqtt_status_enabled) ? "on" : "off"; - if (s_mqtt_bridge_instance == nullptr || !s_mqtt_bridge_instance->_initialized) { + const RuntimeSnapshot snapshot = _published.read(); + if (!snapshot.running) { snprintf(buf, bufsize, "> msgs: %s (bridge %s)", msgs, s_stop_unproven ? "stopped: waiting for the previous stop to finish; reboot if it persists" : "not running"); return; } - MQTTBridge* b = s_mqtt_bridge_instance; - // Build per-slot status strings (compact format to fit 160-byte reply buffer) - // Only show configured slots, skip "none" slots - int q = 0; -#ifdef ESP_PLATFORM - if (b->_packet_queue_handle != nullptr) { - q = (int)uxQueueMessagesWaiting(b->_packet_queue_handle); - } -#else - q = b->_queue_count; -#endif + const int q = snapshot.queue; // replyAppendf clamps pos into the buffer on every call, so no per-append // guard or trailing clamp is needed (see MQTTReplyFormat.h / A1). int pos = 0; replyAppendf(buf, bufsize, &pos, "> msgs: %s", msgs); for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - const MQTTSlot& slot = b->_slots[i]; - const char* name = nullptr; - const char* state = nullptr; - - if (!slot.enabled && slot.preset) { - name = slot.preset->name; - state = "inactive"; - } else if (!slot.enabled) { - continue; // Skip unconfigured slots - } else if (!b->isSlotReady(i)) { - name = slot.preset ? slot.preset->name : "custom"; - state = "wait"; - } else if (slot.connected) { - name = slot.preset ? slot.preset->name : "custom"; - state = "ok"; - } else if (slot.circuit_breaker_tripped) { - name = slot.preset ? slot.preset->name : "custom"; - state = "fail"; - } else { - name = slot.preset ? slot.preset->name : "custom"; - state = "disc"; - } + const SlotRuntimeSnapshot& slot = snapshot.slots[i]; + if (!slot.configured) continue; + const char* name = slot.name; + const char* state = slot.state; replyAppendf(buf, bufsize, &pos, ", %d: %s (%s)", i + 1, name, state); } replyAppendf(buf, bufsize, &pos, ", q:%d", q); @@ -310,16 +332,16 @@ void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPref // Periodic neighbors: time to next publish + how the last one went. if (obs && obs->mqtt_neighbors_enabled) { char when[16]; - switch (b->_neighbors_phase.load(std::memory_order_relaxed)) { + switch (snapshot.neighbors_phase) { case NBR_ACTIVE: strcpy(when, "active"); break; case NBR_DUE: strcpy(when, "due"); break; default: formatDuration(when, sizeof(when), - b->_neighbors_secs_until_next.load(std::memory_order_relaxed)); + snapshot.neighbors_seconds); break; } const char* last; - switch (b->_neighbors_last_result.load(std::memory_order_relaxed)) { + switch (snapshot.neighbors_result) { case NBR_RESULT_OK: last = "ok"; break; case NBR_RESULT_FAIL: last = "failed"; break; default: last = "none"; break; @@ -337,48 +359,33 @@ void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPref // broker whose uplink is dropping writes. void MQTTBridge::formatMqttStatsReply(char* buf, size_t bufsize) { if (buf == nullptr || bufsize == 0) return; - if (s_mqtt_bridge_instance == nullptr || !s_mqtt_bridge_instance->_initialized) { + const RuntimeSnapshot snapshot = _published.read(); + if (!snapshot.running) { snprintf(buf, bufsize, "> (bridge not running)"); return; } - MQTTBridge* b = s_mqtt_bridge_instance; - - int q = 0; -#ifdef ESP_PLATFORM - if (b->_packet_queue_handle != nullptr) { - q = (int)uxQueueMessagesWaiting(b->_packet_queue_handle); - } -#else - q = b->_queue_count; -#endif - - size_t outbox_total = 0; - unsigned long outbox_drops = 0; - for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - if (b->_slots[i].client) { - outbox_total += b->_slots[i].client->getOutboxSize(); - outbox_drops += b->_slots[i].client->getOutboxDrops(); - } - } + const int q = snapshot.queue; + const unsigned long outbox_total = snapshot.outbox; + const unsigned long outbox_drops = snapshot.outbox_drops; // drops=/: outbox-cap drops vs. memory-pressure skips. int pos = 0; replyAppendf(buf, bufsize, &pos, "> Free=%d Max=%d q:%d/%d Outbox=%u drops=%lu/%d", - (int)ESP.getFreeHeap(), (int)ESP.getMaxAllocHeap(), + (int)snapshot.free_heap, (int)snapshot.max_heap, q, MAX_QUEUE_SIZE, (unsigned)outbox_total, - outbox_drops, b->_skipped_publishes); + outbox_drops, snapshot.skipped); // filt=: packets the per-slot type filters rejected before the queue. // Omitted while zero so an unfiltered node's reply keeps its former length — // the per-slot list below is what usually gets clamped away first. - if (b->_filtered_packets > 0) { - replyAppendf(buf, bufsize, &pos, " filt=%lu", b->_filtered_packets); + if (snapshot.filtered > 0) { + replyAppendf(buf, bufsize, &pos, " filt=%lu", snapshot.filtered); } replyAppendf(buf, bufsize, &pos, " |"); for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - if (!b->_slots[i].enabled || !b->_slots[i].client) continue; + if (!snapshot.slots[i].enabled || !snapshot.slots[i].client) continue; replyAppendf(buf, bufsize, &pos, " s%d=%lu/%lu", i + 1, - b->_slots[i].client->getPublishOk(), - b->_slots[i].client->getPublishErr()); + snapshot.slots[i].publish_ok, + snapshot.slots[i].publish_err); } } @@ -387,35 +394,22 @@ void MQTTBridge::formatMqttStatsReply(char* buf, size_t bufsize) { // unconfigured, or bridge-not-running slots. bool MQTTBridge::getSlotStatusSnapshot(int slot_index, SlotStatusSnapshot* out) { if (out == nullptr || slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return false; - if (s_mqtt_bridge_instance == nullptr || !s_mqtt_bridge_instance->_initialized) return false; - MQTTBridge* b = s_mqtt_bridge_instance; - const MQTTSlot& slot = b->_slots[slot_index]; - - if (!slot.enabled && slot.preset) { - out->name = slot.preset->name; - out->state = "inactive"; - } else if (!slot.enabled) { - return false; // unconfigured slot - } else { - out->name = slot.preset ? slot.preset->name : "custom"; - if (!b->isSlotReady(slot_index)) { - out->state = "wait"; - } else if (slot.connected) { - out->state = "ok"; - } else if (slot.circuit_breaker_tripped) { - out->state = "fail"; - } else { - out->state = "disc"; - } - } - out->publish_ok = slot.client ? slot.client->getPublishOk() : 0; - out->publish_err = slot.client ? slot.client->getPublishErr() : 0; - // Lets the portal show why a healthy slot is quiet. - out->filter_mask = b->_obs ? b->_obs->mqtt_slot_packet_filter[slot_index] - : MQTTPacketFilter::kAllPacketTypes; + const RuntimeSnapshot snapshot = _published.read(); + const SlotRuntimeSnapshot& slot = snapshot.slots[slot_index]; + if (!snapshot.running || !slot.configured) return false; + memcpy(out->name, slot.name, sizeof(out->name)); + memcpy(out->state, slot.state, sizeof(out->state)); + out->publish_ok = slot.publish_ok; + out->publish_err = slot.publish_err; + out->filter_mask = slot.filter_mask; return true; } +void MQTTBridge::appendSlotStatsJson(char* buf, size_t size, int pos) { + const RuntimeSnapshot snapshot = _published.read(); + appendObserverSlotStatsJson(buf, size, pos, snapshot.slots, snapshot.running ? RUNTIME_MQTT_SLOTS : 0); +} + int MQTTBridge::getMaxActiveSlots() { // Each WSS/TLS connection needs ~40KB for mbedTLS buffers. Without PSRAM even // 3 concurrent connections would exhaust internal heap, so cap at 2; with @@ -444,21 +438,43 @@ unsigned long MQTTBridge::getLastWifiDisconnectTime() { return activeNetworkLink().lastDisconnectTime(); } +MQTTBridge::SlotOutageSnapshot MQTTBridge::getSlotOutageSnapshot(int slot_index) const { + SlotOutageSnapshot out; + if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return out; + const RuntimeSnapshot snapshot = _published.read(); + const SlotRuntimeSnapshot& slot = snapshot.slots[slot_index]; + out.monitored = snapshot.running && slot.enabled && slot.attempted; + out.started_ms = slot.current_outage_started_ms; + memcpy(out.name, slot.name, sizeof(out.name)); + return out; +} + +void MQTTBridge::formatRuntimeReply(char* buf, size_t bufsize) { + if (!buf || !bufsize) return; + const RuntimeSnapshot s = _published.read(); + snprintf(buf, bufsize, "> %s seq:%lu age:%lums cfg:%lu/%lu events_lost:%lu", + s.running ? "running" : s_stop_unproven ? "stop unproven" : "stopped", + (unsigned long)s.sequence, (unsigned long)((uint32_t)(millis() - s.sampled_ms)), + (unsigned long)s.desired_revision, (unsigned long)s.applied_revision, + (unsigned long)s.event_overflows); +} + unsigned long MQTTBridge::getSlotCurrentOutageStartMs(int slot_index) const { if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return 0; - return _slots[slot_index].current_outage_started_ms; + return _published.read().slots[slot_index].current_outage_started_ms; } bool MQTTBridge::isSlotEnabledAndAttempted(int slot_index) const { if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return false; - const MQTTSlot& s = _slots[slot_index]; - return s.enabled && s.initial_connect_done; + const SlotRuntimeSnapshot s = _published.read().slots[slot_index]; + return s.enabled && s.attempted; } const char* MQTTBridge::getSlotPresetName(int slot_index) const { if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return "?"; - const MQTTSlot& s = _slots[slot_index]; - if (s.preset && s.preset->name) return s.preset->name; + const SlotRuntimeSnapshot s = _published.read().slots[slot_index]; + const MQTTPresetDef* preset = findMQTTPreset(s.name); + if (preset) return preset->name; if (!s.enabled) return MQTT_PRESET_NONE; return MQTT_PRESET_CUSTOM; } @@ -513,7 +529,8 @@ const char* MQTTBridge::tlsErrorStr(int32_t err) { void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) { if (!buf || bufsize == 0) return; - if (!s_mqtt_bridge_instance || !s_mqtt_bridge_instance->_initialized) { + const RuntimeSnapshot snapshot = _published.read(); + if (!snapshot.running) { snprintf(buf, bufsize, "> mqtt%d: bridge not running", slot_index + 1); return; } @@ -522,17 +539,16 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) return; } - MQTTBridge* b = s_mqtt_bridge_instance; - const MQTTSlot& slot = b->_slots[slot_index]; + const SlotRuntimeSnapshot& slot = snapshot.slots[slot_index]; // Determine state string const char* state; - if (!slot.enabled && !slot.preset && slot.host[0] == '\0') { + if (!slot.configured) { snprintf(buf, bufsize, "> mqtt%d: not configured", slot_index + 1); return; } else if (!slot.enabled) { state = "inactive"; - } else if (!b->isSlotReady(slot_index)) { + } else if (!slot.ready) { // Same classification as `get mqtt.status` and getSlotStatusSnapshot(): the slot // is configured but missing a token/IATA/credential, so it was never set up and // has no client yet. Previously reported "disc", which read as a network fault. @@ -560,8 +576,7 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index) // the 160-byte reply, and an error tail (TLS + mbedTLS + errno + age) can // already reach ~117 chars, so putting the filter first would push the // operator's diagnostic detail off the end of a failing slot's line. - const uint16_t filter_mask = b->_obs ? b->_obs->mqtt_slot_packet_filter[slot_index] - : MQTTPacketFilter::kAllPacketTypes; + const uint16_t filter_mask = slot.filter_mask; char filter_text[MQTTPacketFilter::kFilterTextSize]; const bool show_filter = filter_mask != MQTTPacketFilter::kAllPacketTypes && MQTTPacketFilter::format(filter_mask, filter_text, sizeof(filter_text)); @@ -674,13 +689,11 @@ static inline uint32_t mqttStopTimeoutForSlots(int slots) { // --------------------------------------------------------------------------- MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mgr, mesh::RTCClock *rtc, mesh::LocalIdentity *identity) : BridgeBase(prefs, mgr, rtc), - _obs(obs), + _source_obs(obs), _network(&activeNetworkLink()), _queue_count(0), _last_status_publish(0), _last_status_retry(0), _status_interval(300000), - _last_ntp_sync(0), _ntp_synced(false), _ntp_sync_pending(false), _slots_setup_done(false), _max_active_slots(RUNTIME_MQTT_SLOTS), - _ntp_force_requested(false), _ntp_force_done(false), _ntp_force_result(false), - _ntp_diag_requested(false), _ntp_diag_done(false), _ntp_diag_count(0), + _ntp_synced(false), _slots_setup_done(false), _max_active_slots(RUNTIME_MQTT_SLOTS), // Default to UTC; setRules() will be called from syncTimeWithNTP when a // non-UTC timezone string is configured. Timezone has no default ctor, // so we must pass rules here. @@ -749,20 +762,14 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg _slots[i].generation = 0; _slots[i].applied_config = MqttEffectiveConfig(); _slots[i].allocated_buffer_size = 0; - _slot_reconfigure_pending[i] = false; _slot_force_jwt_mint[i] = false; _status_publish_pending[i] = false; } // Reset CLI-requested forced NTP sync handshake (bridge object is reused across restarts) _ntp_force_requested = false; - _ntp_force_done = false; - _ntp_force_result = false; // Reset CLI-requested NTP diagnostic handshake - _ntp_diag_requested = false; - _ntp_diag_done = false; - _ntp_diag_count = 0; #if defined(WITH_MQTT_NEIGHBORS) // Neighbors publish handoff (buffer allocated in begin() after PSRAM probe). @@ -876,6 +883,10 @@ void MQTTBridge::begin() { return; } + _ntp_schedule = NtpSchedule(); + _slots_setup_done = false; + _ntp_force_requested.store(false); + // PSRAM diagnostic - helps debug memory fragmentation on boards with external RAM #ifdef BOARD_HAS_PSRAM { @@ -913,6 +924,10 @@ void MQTTBridge::begin() { MQTT_DEBUG_PRINTLN("Max active slots: %d", _max_active_slots); // Ethernet needs no credentials; Wi-Fi preserves the existing SSID gate. + publishConfiguration(); + uint32_t initial_reconfigure = 0; + _applied_revision = 0; + _configuration.consume(_runtime_prefs, _radio_metadata, _applied_revision, initial_reconfigure); if (!isNetworkConfigValid(_obs)) { MQTT_DEBUG_PRINTLN("MQTT Bridge initialization skipped - %s is not configured", _network->mediumName()); @@ -1076,6 +1091,7 @@ void MQTTBridge::begin() { _stop_requested.store(false, std::memory_order_relaxed); _stop_acked.store(false, std::memory_order_relaxed); _teardown_complete.store(false, std::memory_order_relaxed); + publishRuntimeSnapshot(); BaseType_t create_result = xTaskCreatePinnedToCore( mqttTask, "MQTTBridge", @@ -1088,6 +1104,7 @@ void MQTTBridge::begin() { if (create_result != pdPASS) _mqtt_task_handle = nullptr; if (_mqtt_task_handle == nullptr) { MQTT_DEBUG_PRINTLN("Failed to create MQTT task!"); + _published.publish(RuntimeSnapshot{}); vQueueDelete(_packet_queue_handle); _packet_queue_handle = nullptr; #if defined(BOARD_HAS_PSRAM) @@ -1108,9 +1125,8 @@ void MQTTBridge::begin() { // MQTT client objects are NOT allocated here. setupSlot() creates one on a slot's // first setup, so unconfigured and capped-off slots never cost their ~1.3 KB of - // internal DRAM. Once created a client lives for the bridge's lifetime, so the - // reconfigure/reconnect paths still reuse the same mbedTLS context instead of - // churning ~40 KB of internal heap per cycle. + // internal DRAM. Reconnects reuse the wrapper and SDK task where possible; + // closing a TLS transport still releases its session buffers. // Sync the lifecycle Coordinator to Running now that all resources exist and // the task is created. Driven only on the success path: the failure rollbacks @@ -1121,7 +1137,6 @@ void MQTTBridge::begin() { _lifecycle.onTaskStarted(); // Starting -> Running _initialized = true; - s_mqtt_bridge_instance = this; MQTT_DEBUG_PRINTLN("MQTT Bridge initialized"); } @@ -1130,6 +1145,7 @@ void MQTTBridge::begin() { // --------------------------------------------------------------------------- void MQTTBridge::end() { MQTT_DEBUG_PRINTLN("Stopping MQTT Bridge..."); + _ntp_diag_job.cancel(millis()); // Idempotent stop: nothing to tear down if we never started (or already stopped). if (!_initialized) { @@ -1137,8 +1153,10 @@ void MQTTBridge::end() { return; } - // Stop new diagnostic reads through the singleton before teardown begins. - s_mqtt_bridge_instance = nullptr; + // Readers see a stopped snapshot while teardown proceeds. + RuntimeSnapshot stopped = _published.read(); + stopped.running = false; + _published.publish(stopped); // Size the stop timeout to the work about to happen: each enabled slot's // mbedTLS/wss client takes ~5-6 s to disconnect + destroy, sequentially (Phase @@ -1192,7 +1210,10 @@ void MQTTBridge::end() { // above). begin() refuses until the task acknowledges, so _initialized == false // cannot be turned into a second task over the same state. _initialized = false; - _slots_setup_done = false; // Reset so deferred setup runs again on next begin() + RuntimeSnapshot stopped_snapshot = _published.read(); + stopped_snapshot.running = false; + stopped_snapshot.connected = 0; + _published.publish(stopped_snapshot); s_stop_unproven = _lifecycle.isStopUnproven(); if (_lifecycle.isStopUnproven()) { MQTT_DEBUG_PRINTLN("MQTT Bridge stop UNPROVEN after %lu ms: task did not acknowledge. " @@ -1322,8 +1343,7 @@ void MQTTBridge::initializeNetworkInTask() { // not race the first DNS lookup after a bridge restart. if (!_network->begin(_obs->wifi_ssid, _obs->wifi_password)) { MQTT_DEBUG_PRINTLN("%s network initialization failed", _network->mediumName()); - } else if (_network->isConnected() && !_ntp_synced && !_ntp_sync_pending) { - _ntp_sync_pending = true; + } else if (_network->isConnected() && !_ntp_synced) { } // NOTE: Slot setup is deferred until after NTP sync in mqttTaskLoop(). @@ -1346,8 +1366,6 @@ void MQTTBridge::mqttTaskLoop() { // Treat the first post-settle sample as an edge. A fast Wi-Fi association // may complete during the delay above, before the task has sampled state; // initializing from the live value would then defer NTP until its retry. - bool network_was_connected = false; - unsigned long last_ntp_attempt = 0; // Main task loop #ifdef MQTT_MEMORY_DEBUG @@ -1366,6 +1384,7 @@ void MQTTBridge::mqttTaskLoop() { teardownSlot(i); } destroySlotClients(); + publishRuntimeSnapshot(); // The acknowledgement means "every client is destroyed, nothing I own is // still running". One client whose esp_mqtt_client_stop() never completed @@ -1428,48 +1447,18 @@ void MQTTBridge::mqttTaskLoop() { } #endif + processSlotEvents(); + applyConfiguration(); handleNetworkConnection(now); - // Schedule once per link-up edge. Failed syncs are owned by the 30-second - // retry below; re-arming here on every loop would make a blocked NTP path - // run the multi-second sync sequence back-to-back forever. - const bool network_connected = _network->isConnected(); - _ntp_sync_pending = NetworkPolicy::ntpPendingAfterConnectivitySample( - _ntp_synced, _ntp_sync_pending, network_was_connected, - network_connected); - network_was_connected = network_connected; - if (_ntp_sync_pending && network_connected) { - _ntp_sync_pending = false; - last_ntp_attempt = now; - syncTimeWithNTP(); - } - - // Retry NTP every 30s if initial sync failed (slots can't start without valid time) - if (NetworkPolicy::ntpRetryDue( - _ntp_synced, network_connected, (uint32_t)now, - (uint32_t)last_ntp_attempt)) { - last_ntp_attempt = now; - syncTimeWithNTP(); - } - - // Process a CLI-requested forced NTP sync (queued from Core 1). Running it here - // keeps all NTP I/O on Core 0; requestForcedNtpSync() blocks the CLI thread until - // we publish the outcome below. - if (_ntp_force_requested) { - _ntp_force_requested = false; - // primary_only: validate just the server that was set, so a typo fails fast. - bool ok = syncTimeWithNTP(true, /*primary_only=*/true); - _ntp_force_result = ok; - _ntp_force_done = true; // set last so the waiter sees a consistent result - } - - // Process a CLI-requested NTP connectivity diagnostic (queued from Core 1). - // Probe-only — never touches the system clock. - if (_ntp_diag_requested) { - _ntp_diag_requested = false; - runNtpDiagProbe(); - _ntp_diag_done = true; // set last so the waiter sees populated results + const bool ntp_due = _ntp_schedule.due((uint32_t)millis()); + if (_ntp_force_requested.exchange(false, std::memory_order_acq_rel)) { + syncTimeWithNTP(true, /*primary_only=*/true); + } else if (ntp_due && _network->isConnected()) { + syncTimeWithNTP(false, false, _ntp_synced ? 1 : 2); } + const uint32_t ntp_probe = _ntp_diag_job.begin(); + if (ntp_probe) runNtpDiagProbe(ntp_probe); // 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. @@ -1509,14 +1498,7 @@ void MQTTBridge::mqttTaskLoop() { } } - // Process pending slot reconfigures (queued from CLI on Core 1) - for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - 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]); - } - } + processSlotEvents(); // 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 @@ -1578,13 +1560,6 @@ void MQTTBridge::mqttTaskLoop() { // Periodic configuration check (throttled to avoid spam) checkConfigurationMismatch(); - // 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 (_network->isConnected() && now - _last_ntp_sync > 3600000) { - refreshNTP(); - } - // Publish status updates (handle millis() overflow correctly). // Read the toggle live from prefs (like mqtt.packets/rx/tx below) so a // CLI/web `set mqtt.status` change applies without a bridge restart. @@ -1650,6 +1625,8 @@ void MQTTBridge::mqttTaskLoop() { last_slot_status_update = now; } + publishRuntimeSnapshot(); + // Adaptive delay: 5 ms when packets are queued, 50 ms when idle. // The previous "status approaching" check (widening to 5 ms for 10 s before each status // publish) caused 2 000 unnecessary wakeups per interval; the 50 ms idle tick catches @@ -1670,11 +1647,8 @@ void MQTTBridge::mqttTaskLoop() { // at all until setupSlot() runs (the reconnect ladder is gated on // initial_connect_done), so slots that are unconfigured or capped off never get one. // -// Once created the object lives until destroySlotClients(): reconfiguring a slot -// (preset change, JWT renewal, reconnect) reuses it, so the mbedTLS context and its -// ~40 KB of internal-heap buffers are allocated once instead of every reconfigure. -// That context is created by connect(), not by this constructor, so deferring the -// allocation to first use costs nothing beyond the object itself. +// Compatible reconnects reuse the wrapper. Transport/trust changes can recreate +// it after a proven stop. TLS session buffers still follow connection lifetime. bool MQTTBridge::ensureSlotClient(int index) { if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; MQTTSlot& slot = _slots[index]; @@ -1696,98 +1670,63 @@ bool MQTTBridge::ensureSlotClient(int index) { slot.allocated_buffer_size = kMqttClientBufferSize; slot.client->setAutoReconnect(false); // we handle reconnect with our own backoff - slot.client->onConnect([this, index](bool sessionPresent) { - // A CONNECT started before this slot was disabled or reconfigured can still - // complete afterwards. Accepting it marked a slot connected that the - // operator had switched off, scheduled its status publish, and published - // through the old session (F04). The bridge task owns client_state, so it - // is the authority on whether this event was asked for. - const ClientState st = _slots[index].client_state; - if (!_slots[index].enabled || !(st == ClientState::Starting || st == ClientState::Disconnected)) { - MQTT_DEBUG_PRINTLN("MQTT%d ignoring late CONNECTED (state=%s, gen=%lu, enabled=%d)", - index + 1, clientStateName(st), - (unsigned long)_slots[index].generation, (int)_slots[index].enabled); - return; - } - MQTT_DEBUG_PRINTLN("MQTT%d connected", index + 1); - _slots[index].client_state = ClientState::Connected; - _slots[index].connected = true; - _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 - // accepts and then drops within seconds would reset the ladder every - // cycle and retry at the 10 s rung forever, and each retry is a full - // TLS session alloc/free (~40 KB of internal-heap churn, a known - // fragmentation driver). The ladder is instead cleared by - // maintainSlotConnection() once the connection has stayed up for - // BACKOFF_STABLE_RESET_MS, so flapping endpoints keep their earned - // backoff level. The breaker itself does clear now: while connected - // the diag/status must not claim the slot gave up, and the next - // disconnect should be governed by the (still-elevated) ladder. - _slots[index].connected_at_ms = millis(); - _slots[index].circuit_breaker_tripped = false; - _slots[index].last_tls_err = 0; - _slots[index].last_tls_stack_err = 0; - _slots[index].last_sock_errno = 0; - _slots[index].last_connack_code = 0; - _slots[index].last_error_time = 0; - _slots[index].current_outage_started_ms = 0; // clear current-outage timer for AlertReporter - updateCachedConnectionStatus(); // bool store — safe from this (esp-mqtt) task - // This callback runs on the client's esp-mqtt event task, not the bridge - // task. Do NOT build/publish status here: publishStatusToSlot() writes the - // shared _json_scratch_doc/_json_scratch_buffer/_origin that the periodic - // publishStatus() uses on the bridge task, and two slots' callbacks could - // race each other over them. Marshal the publish onto the bridge task via a - // per-slot flag (see mqttTaskLoop consumer / A2). - _status_publish_pending[index] = true; + const uint32_t incarnation = ++slot.incarnation; + slot.client->onConnect([this, index, incarnation](bool) { + MqttSlotEvent event; + event.kind = MqttSlotEvent::Kind::Connected; + event.slot = index; + event.incarnation = incarnation; + event.at_ms = millis(); + _slot_events.push(event); }); - slot.client->onDisconnect([this, index](bool sessionPresent) { - MQTT_DEBUG_PRINTLN("MQTT%d disconnected", index + 1); - // Only a live client's disconnect is news. One arriving for a client we - // already stopped (or quarantined) must not resurrect its state. - if (clientStateIsLive(_slots[index].client_state)) { - _slots[index].client_state = ClientState::Disconnected; - } - _slots[index].disconnect_count++; - if (_slots[index].first_disconnect_time == 0) { - _slots[index].first_disconnect_time = millis(); - } - if (_slots[index].current_outage_started_ms == 0) { - _slots[index].current_outage_started_ms = millis(); - } - _slots[index].connected = false; - _slots[index].connected_at_ms = 0; // stability clock only runs while connected - updateCachedConnectionStatus(); + slot.client->onDisconnect([this, index, incarnation](bool) { + MqttSlotEvent event; + event.kind = MqttSlotEvent::Kind::Disconnected; + event.slot = index; + event.incarnation = incarnation; + event.at_ms = millis(); + _slot_events.push(event); }); - slot.client->onError([this, index](esp_mqtt_error_codes error) { - _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; - _slots[index].last_error_time = millis(); - // Cleared on any other error type so the diag describes the latest failure - // rather than pairing a fresh transport error with an old refusal. - _slots[index].last_connack_code = - (error.error_type == MQTT_ERROR_TYPE_CONNECTION_REFUSED) - ? (uint8_t)error.connect_return_code : 0; - if (error.error_type == MQTT_ERROR_TYPE_CONNECTION_REFUSED) { - _slot_force_jwt_mint[index] = true; - // Broker rejected the MQTT CONNECT itself — not a transport failure. - // return code: 1=protocol, 2=client-id rejected, 3=server unavailable, - // 4=bad username/password, 5=not authorized. Codes 3/4/5 point at a - // server-side lockout or auth problem rather than the network. - MQTT_DEBUG_PRINTLN("MQTT%d connection refused by broker (return code=%d)", - index + 1, (int)error.connect_return_code); - } else if (error.esp_tls_last_esp_err != 0 || error.esp_tls_stack_err != 0 || error.esp_transport_sock_errno != 0) { - MQTT_DEBUG_PRINTLN("MQTT%d error: tls=%d, tls_stack=%d, sock=%d, type=%d", - index + 1, error.esp_tls_last_esp_err, error.esp_tls_stack_err, - error.esp_transport_sock_errno, error.error_type); - } else { - MQTT_DEBUG_PRINTLN("MQTT%d error: type=%d", index + 1, error.error_type); - } + slot.client->onError([this, index, incarnation](esp_mqtt_error_codes error) { + MqttSlotEvent event; + event.kind = MqttSlotEvent::Kind::Error; + event.slot = index; + event.incarnation = incarnation; + event.at_ms = millis(); + event.tls = error.esp_tls_last_esp_err; + event.tls_stack = error.esp_tls_stack_err; + event.socket = error.esp_transport_sock_errno; + event.connack = error.error_type == MQTT_ERROR_TYPE_CONNECTION_REFUSED + ? static_cast(error.connect_return_code) : 0; + _slot_events.push(event); }); return true; } +void MQTTBridge::processSlotEvents() { + const uint32_t overflow = _slot_events.takeOverflowSlots(); + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { + if (!(overflow & (uint32_t(1) << i))) continue; + // Losing lifecycle events makes connectivity unproven. Stop before allowing + // another publish, then let ordinary maintenance restart the joined client. + _slots[i].connected = false; + _status_publish_pending[i] = false; + if (_slots[i].client && clientStateIsLive(_slots[i].client_state)) stopSlotClient(i); + _slot_events.discardSlot(i); + } + MqttSlotEvent event; + for (unsigned n = 0; n < 32 && _slot_events.pop(event); ++n) { + const int index = event.slot; + if (index >= RUNTIME_MQTT_SLOTS) continue; + if (applyMqttSlotEvent(_slots[index], event, _slot_force_jwt_mint[index], _status_publish_pending[index])) { + MQTT_DEBUG_PRINTLN("MQTT%d event=%u state=%s incarnation=%lu connack=%u tls=%ld sock=%ld", + index + 1, (unsigned)event.kind, clientStateName(_slots[index].client_state), + (unsigned long)event.incarnation, (unsigned)event.connack, (long)event.tls, (long)event.socket); + } + } + updateCachedConnectionStatus(); +} + // Allocate this slot's JWT token buffer. Called only from createSlotAuthToken(), the // sole writer, so a slot on a non-JWT preset (or no preset at all) never allocates. // @@ -2292,10 +2231,6 @@ bool MQTTBridge::recreateSlotClient(int index) { return ensureSlotClient(index); } -// Disconnect the slot's MQTT client and clear per-connection state, but leave -// the client object alive so a subsequent setupSlot() can reuse its mbedTLS -// context. This is called both on reconfigure (preset change) and at shutdown; -// destruction of the underlying client happens once in destroySlotClients(). // Close a live client for a reconfigure: cheap where that is safe, a real stop // where it is not. // @@ -2326,7 +2261,13 @@ void MQTTBridge::closeLiveClientForReconfigure(int index) { MQTT_DEBUG_PRINTLN("MQTT%d reconfigure: transport did not close cleanly (%s)", index + 1, esp_err_to_name(r)); } + if (r != ESP_OK) { + stopSlotClient(index); + return; + } + _slot_events.discardSlot(index); slot.client_state = ClientState::Disconnected; + slot.connected = false; } // The old route is already gone, so a link transition waits far less for the @@ -2413,6 +2354,8 @@ void MQTTBridge::stopSlotClient(int index) { MQTT_DEBUG_PRINTLN("MQTT%d stop: client was not started (nothing to join)", index + 1); } slot.client_state = ClientState::Stopped; + slot.connected = false; + _slot_events.discardSlot(index); return; } MQTT_DEBUG_PRINTLN("MQTT%d stop returned %s - client quarantined for this boot", @@ -2479,21 +2422,9 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) { return ESP_ERR_INVALID_STATE; } - esp_err_t r; - if (!slot.client->isStarted()) { - MQTT_DEBUG_PRINTLN("MQTT%d start (client was stopped)", index + 1); - r = slot.client->connect(); - } else { - r = slot.client->reconnect(); - } - if (r == ESP_OK) { - slot.client_state = ClientState::Starting; - slot.generation++; - // The attempt carries whatever credential is configured right now, so this - // is the point at which a freshly minted token becomes the one in use. - slot.applied_token_expires_at = slot.token_expires_at; - } - return r; + return startMqttSlotAttempt(slot, [&]() { + return slot.client->isStarted() ? slot.client->reconnect() : slot.client->connect(); + }); } @@ -2703,8 +2634,8 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns MQTT_DEBUG_PRINTLN("MQTT%d int_heap=%d at token renewal reconnect", index + 1, (int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL)); MQTT_DEBUG_PRINTLN(" radio_state=%d, last_rx=%lums ago", - _radio ? _radio->getRadioState() : -1, - (_radio && _radio->getLastRecvMillis() > 0) ? (_ms->getMillis() - _radio->getLastRecvMillis()) : 0); + _radio_stats.read().state, + _radio_stats.read().last_rx_age_ms); } else { // Token renewed but old one still valid — just update credentials for next reconnect slot.client->setCredentials(_jwt_username, slot.auth_token); @@ -2721,11 +2652,6 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns } } - // Phase 4 (MQTT memory-defrag): the MIN_TLS_HEAP preflight was a workaround - // for the fragmentation caused by per-reconnect mbedTLS allocations. With - // persistent clients (Phase 1), the mbedTLS context is allocated once at - // startup and the preflight is no longer necessary. - const auto prepareJwtReconnect = [&](bool force_mint, int backoff_level) { const bool has_token = slot.auth_token && slot.auth_token[0] != '\0'; const unsigned long expires_at = slot.token_expires_at; @@ -2800,8 +2726,8 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker probe (attempting single reconnect after %lu ms, int_heap=%d)", index + 1, probe_elapsed, (int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL)); MQTT_DEBUG_PRINTLN(" radio_state=%d, last_rx=%lums ago", - _radio ? _radio->getRadioState() : -1, - (_radio && _radio->getLastRecvMillis() > 0) ? (_ms->getMillis() - _radio->getLastRecvMillis()) : 0); + _radio_stats.read().state, + _radio_stats.read().last_rx_age_ms); if (slot_uses_jwt) { prepareJwtReconnect(true, -1); } @@ -2835,8 +2761,8 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns MQTT_DEBUG_PRINTLN("MQTT%d reconnecting (backoff level %d, failures at max: %d, int_heap=%d)", index + 1, slot.reconnect_backoff, slot.max_backoff_failures, (int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL)); MQTT_DEBUG_PRINTLN(" radio_state=%d, last_rx=%lums ago", - _radio ? _radio->getRadioState() : -1, - (_radio && _radio->getLastRecvMillis() > 0) ? (_ms->getMillis() - _radio->getLastRecvMillis()) : 0); + _radio_stats.read().state, + _radio_stats.read().last_rx_age_ms); reconnect_attempted = true; _last_slot_reconnect_ms = now_millis; if (slot_uses_jwt) { @@ -3092,7 +3018,7 @@ void MQTTBridge::publishStatusToSlot(int index) { MQTTMessageBuilder::formatIsoTimestampForMqtt(now_tv.tv_sec, now_tv.tv_usec, _timezone, timestamp, sizeof(timestamp)); snprintf(radio_info, sizeof(radio_info), "%.6f,%.1f,%d,%d", - _prefs->freq, _prefs->bw, _prefs->sf, _prefs->cr); + _radio_metadata.freq, _radio_metadata.bw, _radio_metadata.sf, _radio_metadata.cr); strncpy(origin_id, _device_id, sizeof(origin_id) - 1); origin_id[sizeof(origin_id) - 1] = '\0'; @@ -3100,30 +3026,16 @@ void MQTTBridge::publishStatusToSlot(int index) { char client_version[64]; getClientVersion(client_version, sizeof(client_version)); - // Collect stats on-demand if sources are available - int battery_mv = -1; - int uptime_secs = -1; - int errors = -1; - int noise_floor = -999; - int tx_air_secs = -1; - int rx_air_secs = -1; - int recv_errors = -1; - int packets_sent = -1; - int packets_received = -1; - - if (_board) battery_mv = _board->getBattMilliVolts(); - if (_ms) uptime_secs = _ms->getMillis() / 1000; - if (_dispatcher) { - errors = _dispatcher->getErrFlags(); - tx_air_secs = _dispatcher->getTotalAirTime() / 1000; - rx_air_secs = _dispatcher->getReceiveAirTime() / 1000; - packets_sent = (int)(_dispatcher->getNumSentFlood() + _dispatcher->getNumSentDirect()); - packets_received = (int)(_dispatcher->getNumRecvFlood() + _dispatcher->getNumRecvDirect()); - } - if (_radio) { - noise_floor = (int16_t)_radio->getNoiseFloor(); - recv_errors = (int)_radio->getPacketsRecvErrors(); - } + const RadioStatsSnapshot stats = _radio_stats.read(); + const int battery_mv = stats.battery_mv; + const int uptime_secs = stats.uptime_secs; + const int errors = stats.errors; + const int noise_floor = stats.noise_floor; + const int tx_air_secs = stats.tx_air_secs; + const int rx_air_secs = stats.rx_air_secs; + const int recv_errors = stats.recv_errors; + const int packets_sent = stats.packets_sent; + const int packets_received = stats.packets_received; // Internal heap free (for diagnosing repeater hangs from internal heap exhaustion) int internal_heap_free = (int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL); @@ -3135,7 +3047,7 @@ void MQTTBridge::publishStatusToSlot(int index) { battery_mv, uptime_secs, errors, _queue_count, noise_floor, tx_air_secs, rx_air_secs, recv_errors, internal_heap_free, packets_sent, packets_received, - _prefs->disable_fwd ? "off" : "on" + _radio_metadata.disable_fwd ? "off" : "on" ); if (len > 0) { @@ -3171,21 +3083,55 @@ bool MQTTBridge::isAnySlotConnected() { return false; } +void MQTTBridge::publishRadioStats() { + RadioStatsSnapshot s; + if (_board) s.battery_mv = _board->getBattMilliVolts(); + if (_ms) s.uptime_secs = _ms->getMillis() / 1000; + if (_dispatcher) { + s.errors = _dispatcher->getErrFlags(); + s.tx_air_secs = _dispatcher->getTotalAirTime() / 1000; + s.rx_air_secs = _dispatcher->getReceiveAirTime() / 1000; + s.packets_sent = _dispatcher->getNumSentFlood() + _dispatcher->getNumSentDirect(); + s.packets_received = _dispatcher->getNumRecvFlood() + _dispatcher->getNumRecvDirect(); + } + if (_radio) { + s.noise_floor = _radio->getNoiseFloor(); + s.recv_errors = _radio->getPacketsRecvErrors(); + s.state = _radio->getRadioState(); + if (_ms && _radio->getLastRecvMillis()) s.last_rx_age_ms = _ms->getMillis() - _radio->getLastRecvMillis(); + } + _radio_stats.publish(s); +} + +void MQTTBridge::publishConfiguration(uint32_t reconfigure) { + RadioMetadata metadata{}; + memcpy(metadata.node_name, _prefs->node_name, sizeof(metadata.node_name)); + metadata.freq = _prefs->freq; + metadata.bw = _prefs->bw; + metadata.sf = _prefs->sf; + metadata.cr = _prefs->cr; + metadata.disable_fwd = _prefs->disable_fwd; + _configuration.publish(*_source_obs, metadata, reconfigure); +} + +void MQTTBridge::applyConfiguration() { + uint32_t reconfigure = 0; + if (!_configuration.consume(_runtime_prefs, _radio_metadata, _applied_revision, reconfigure)) return; + refreshOriginFromPrefs(); + snprintf(_iata, sizeof(_iata), "%s", _obs->mqtt_iata); + _status_interval = _obs->mqtt_status_interval >= 1000 && _obs->mqtt_status_interval <= 3600000 + ? _obs->mqtt_status_interval : 300000; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { + if (reconfigure & (uint32_t(1) << i)) applySlotPreset(i, _obs->mqtt_slot_preset[i]); + } +} + void MQTTBridge::setSlotPreset(int slot_index, const char* preset_name) { 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). - // Set a flag so the MQTT task picks it up on its next loop iteration. - #ifdef ESP_PLATFORM - if (_mqtt_task_handle != nullptr) { - _slot_reconfigure_pending[slot_index] = true; - MQTT_DEBUG_PRINTLN("MQTT%d reconfigure queued (preset: %s)", slot_index + 1, preset_name); - return; - } - #endif - - // Non-ESP32 or bridge not yet started: apply directly - applySlotPreset(slot_index, preset_name); + // The durable image already contains preset_name. Publish it and the request + // together so the worker cannot apply the request to another revision. + (void)preset_name; + publishConfiguration(uint32_t(1) << slot_index); } void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) { @@ -3224,7 +3170,7 @@ void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) { strncpy(slot.audience, _obs->mqtt_slot_audience[slot_index], sizeof(slot.audience) - 1); slot.audience[sizeof(slot.audience) - 1] = '\0'; slot.enabled = (slot.host[0] != '\0'); - if (_initialized && slot.enabled && customEndpointComplete(slot.host, slot.port)) { + if (!_stop_requested.load() && slot.enabled && customEndpointComplete(slot.host, slot.port)) { // Same cap startup applies. teardownSlot() above already released this slot's own // position, so reconfiguring a live slot still passes. if (!canActivateSlot(slot_index)) { @@ -3247,7 +3193,7 @@ void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) { strncpy(slot.password, _obs->mqtt_slot_password[slot_index], sizeof(slot.password) - 1); slot.password[sizeof(slot.password) - 1] = '\0'; } - if (_initialized) { + if (!_stop_requested.load()) { char reason[80]; if (!isSlotReady(slot_index, reason, sizeof(reason))) { MQTT_DEBUG_PRINTLN("MQTT%d (%s) not ready - run '%s' to connect", slot_index + 1, preset_name, reason); @@ -3265,8 +3211,15 @@ void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) { } } +bool MQTTBridge::mayEditStartupMetadata() { + if (_initialized) return false; + pollLateStopAck(); + return !_lifecycle.isStopUnproven(); +} + void MQTTBridge::setSlotCustomBroker(int slot_index, const char* host, uint16_t port, const char* username, const char* password) { + if (!mayEditStartupMetadata()) return; if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return; MQTTSlot& slot = _slots[slot_index]; @@ -3349,7 +3302,7 @@ bool MQTTBridge::handleNetworkConnection(unsigned long now) { } bool MQTTBridge::isReady() const { - return _initialized && isNetworkConfigValid(_obs); + return _initialized && isNetworkConfigValid(_source_obs); } bool MQTTBridge::isIATAValid() const { @@ -3405,16 +3358,22 @@ void MQTTBridge::loop() { if (!_initialized) return; #ifdef ESP_PLATFORM - // On ESP32, loop() is a no-op - all processing happens in the FreeRTOS task + // Publish only between loop-owned commands, after save/rollback has finished. + const uint32_t now = millis(); + if ((uint32_t)(now - _last_radio_sample_ms) >= 1000) { + _last_radio_sample_ms = now; + publishRadioStats(); + } + if ((uint32_t)(now - _last_config_sample_ms) >= 100) { + _last_config_sample_ms = now; + publishConfiguration(); + } return; #else unsigned long now = millis(); - if (handleNetworkConnection(now) && !_ntp_synced) { - syncTimeWithNTP(); - } - if (_ntp_sync_pending && _network->isConnected()) { - _ntp_sync_pending = false; - syncTimeWithNTP(); + handleNetworkConnection(now); + if (_ntp_schedule.due((uint32_t)millis()) && _network->isConnected()) { + syncTimeWithNTP(false, false, _ntp_synced ? 1 : 2); } // Deferred slot setup after NTP sync (non-ESP32 path) @@ -3434,13 +3393,8 @@ void MQTTBridge::loop() { } } - // Process pending slot reconfigures - for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - if (_slot_reconfigure_pending[i]) { - _slot_reconfigure_pending[i] = false; - applySlotPreset(i, _obs->mqtt_slot_preset[i]); - } - } + applyConfiguration(); + processSlotEvents(); // Maintain slot connections (token renewal, reconnect with backoff) maintainSlotConnections(); @@ -3451,11 +3405,6 @@ void MQTTBridge::loop() { // Periodic configuration check (throttled to avoid spam) checkConfigurationMismatch(); - // Periodic NTP refresh (every hour) — lightweight, non-blocking. - if (_network->isConnected() && millis() - _last_ntp_sync > 3600000) { - refreshNTP(); - } - // Publish status updates (handle millis() overflow correctly). // Read the toggle live from prefs so a CLI/web `set mqtt.status` change // applies without a bridge restart. @@ -3523,7 +3472,7 @@ void MQTTBridge::loop() { // --------------------------------------------------------------------------- void MQTTBridge::onPacketReceived(mesh::Packet *packet) { - if (!_initialized || !_obs->mqtt_packets_enabled || !_obs->mqtt_rx_enabled) return; + if (!_initialized || !_source_obs->mqtt_packets_enabled || !_source_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 @@ -3541,8 +3490,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; + uint8_t tx_mode = _source_obs->mqtt_tx_enabled; // Read live from prefs (no restart needed) + if (!_initialized || !_source_obs->mqtt_packets_enabled || tx_mode == 0) return; // Advert mode: only queue self-originated advert packets if (tx_mode == 2) { @@ -3648,6 +3597,7 @@ void MQTTBridge::processPacketQueue() { _last_snr = queued.snr; _last_rssi = queued.rssi; _last_raw_timestamp = millis(); + _last_packet_score = queued.score; } bool packet_eligible = false; @@ -3655,7 +3605,7 @@ void MQTTBridge::processPacketQueue() { packet_eligible, queued.has_raw_data ? queued.raw_data : nullptr, queued.has_raw_data ? queued.raw_len : 0, - queued.snr, queued.rssi); + queued.snr, queued.rssi, queued.score); taskYIELD(); // allow higher-priority tasks to run between packet publishes // Publish raw if enabled (live from prefs so `set mqtt.raw` applies without @@ -3783,6 +3733,7 @@ void MQTTBridge::processPacketQueue() { _last_snr = queued.snr; _last_rssi = queued.rssi; _last_raw_timestamp = millis(); + _last_packet_score = queued.score; } bool packet_eligible = false; @@ -3790,7 +3741,7 @@ void MQTTBridge::processPacketQueue() { packet_eligible, queued.has_raw_data ? queued.raw_data : nullptr, queued.has_raw_data ? queued.raw_len : 0, - queued.snr, queued.rssi); + queued.snr, queued.rssi, queued.score); // No taskYIELD() on non-ESP32 platforms (non-FreeRTOS, cooperative scheduling not needed) // Live from prefs so `set mqtt.raw` applies without a bridge restart. @@ -3900,17 +3851,18 @@ uint8_t MQTTBridge::eligiblePacketSlots(uint8_t packet_type, MQTTMessageType typ // from "no broker configured at all" (the pre-existing silent drop). bool MQTTBridge::shouldQueuePacketType(uint8_t packet_type, bool& filtered) { filtered = false; - if (!_obs) return false; + if (!_source_obs) return false; + const RuntimeSnapshot snapshot = _published.read(); uint16_t masks[RUNTIME_MQTT_SLOTS]; bool enabled[RUNTIME_MQTT_SLOTS]; bool any_enabled = false; for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) { - masks[i] = _obs->mqtt_slot_packet_filter[i]; + masks[i] = _source_obs->mqtt_slot_packet_filter[i]; // Configured, not allocated — see eligiblePacketSlots(). Gating on the client // here would silently drop every packet received before the post-NTP-sync slot // setup, which is exactly the window the queue exists to cover. - enabled[i] = _slots[i].enabled; + enabled[i] = snapshot.slots[i].enabled; any_enabled = any_enabled || enabled[i]; } if (!any_enabled) return false; @@ -3950,7 +3902,7 @@ bool MQTTBridge::publishStatus() { MQTTMessageBuilder::formatIsoTimestampForMqtt(now_tv.tv_sec, now_tv.tv_usec, _timezone, timestamp, sizeof(timestamp)); snprintf(radio_info, sizeof(radio_info), "%.6f,%.1f,%d,%d", - _prefs->freq, _prefs->bw, _prefs->sf, _prefs->cr); + _radio_metadata.freq, _radio_metadata.bw, _radio_metadata.sf, _radio_metadata.cr); strncpy(origin_id, _device_id, sizeof(origin_id) - 1); origin_id[sizeof(origin_id) - 1] = '\0'; @@ -3958,30 +3910,16 @@ bool MQTTBridge::publishStatus() { char client_version[64]; getClientVersion(client_version, sizeof(client_version)); - // Collect stats on-demand if sources are available - int battery_mv = -1; - int uptime_secs = -1; - int errors = -1; - int noise_floor = -999; - int tx_air_secs = -1; - int rx_air_secs = -1; - int recv_errors = -1; - int packets_sent = -1; - int packets_received = -1; - - if (_board) battery_mv = _board->getBattMilliVolts(); - if (_ms) uptime_secs = _ms->getMillis() / 1000; - if (_dispatcher) { - errors = _dispatcher->getErrFlags(); - tx_air_secs = _dispatcher->getTotalAirTime() / 1000; - rx_air_secs = _dispatcher->getReceiveAirTime() / 1000; - packets_sent = (int)(_dispatcher->getNumSentFlood() + _dispatcher->getNumSentDirect()); - packets_received = (int)(_dispatcher->getNumRecvFlood() + _dispatcher->getNumRecvDirect()); - } - if (_radio) { - noise_floor = (int16_t)_radio->getNoiseFloor(); - recv_errors = (int)_radio->getPacketsRecvErrors(); - } + const RadioStatsSnapshot stats = _radio_stats.read(); + const int battery_mv = stats.battery_mv; + const int uptime_secs = stats.uptime_secs; + const int errors = stats.errors; + const int noise_floor = stats.noise_floor; + const int tx_air_secs = stats.tx_air_secs; + const int rx_air_secs = stats.rx_air_secs; + const int recv_errors = stats.recv_errors; + const int packets_sent = stats.packets_sent; + const int packets_received = stats.packets_received; // Internal heap free (for diagnosing repeater hangs from internal heap exhaustion) int internal_heap_free = (int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL); @@ -3993,7 +3931,7 @@ bool MQTTBridge::publishStatus() { battery_mv, uptime_secs, errors, _queue_count, noise_floor, tx_air_secs, rx_air_secs, recv_errors, internal_heap_free, packets_sent, packets_received, - _prefs->disable_fwd ? "off" : "on" + _radio_metadata.disable_fwd ? "off" : "on" ); if (len > 0) { @@ -4025,7 +3963,7 @@ bool MQTTBridge::publishStatus() { bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, bool& has_eligible_target, const uint8_t* raw_data, int raw_len, - float snr, float rssi) { + float snr, float rssi, float captured_score) { has_eligible_target = false; if (!packet) return false; @@ -4040,11 +3978,9 @@ bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, // Memory pressure check: Skip publishes when there's not enough contiguous // heap for the publish itself (JSON buffer + esp-mqtt outbox frame + WiFi TX - // path). Headroom only — NOT an mbedTLS preflight: persistent clients keep - // their TLS contexts allocated for the bridge lifetime, so the old ~52 KB - // "reserve space for reconnect" guard is obsolete post Phase 1. Publish - // payload is capped at PUBLISH_JSON_BUFFER_SIZE (2 KB); 8 KB is a safe - // ceiling including esp-mqtt frame overhead and transient TCP buffers. + // path). This is publish headroom, not a guarantee that a later TLS handshake + // can allocate its record buffers. Payload is capped at 2 KB; the remaining + // headroom covers MQTT frame overhead and transient TCP buffers. #ifdef ESP32 #if defined(BOARD_HAS_PSRAM) static const size_t PUBLISH_SKIP_MAX_ALLOC_THRESHOLD = 16000; @@ -4096,21 +4032,21 @@ bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, origin_id[sizeof(origin_id) - 1] = '\0'; // Firmware rebroadcast "score" for this packet — only meaningful for RX packets - // (depends on receive SNR). Recomputed here exactly as Dispatcher::checkRecv() does, - // via the radio's packetScore(snr, len); NaN signals "omit" (tx, or no radio). + // (depends on receive SNR). Captured on the mesh task using packetScore(snr, len); + // NaN signals "omit" (tx, or no radio). // buildPacketMessage scales it x1000 to match the integer in the serial RX log. // Build packet message using raw radio data if provided int len; if (raw_data && raw_len > 0) { - float score = (_radio && !is_tx) ? _radio->packetScore(snr, raw_len) : NAN; + float score = captured_score; len = MQTTMessageBuilder::buildPacketJSONFromRaw( _json_scratch_doc, raw_data, raw_len, packet, is_tx, _origin, origin_id, snr, rssi, score, _timezone, active_buffer, active_buffer_size ); } else if (!is_tx && _last_raw_data && _last_raw_len > 0 && (millis() - _last_raw_timestamp) < 1000) { - float score = _radio ? _radio->packetScore(_last_snr, _last_raw_len) : NAN; + float score = _last_packet_score; len = MQTTMessageBuilder::buildPacketJSONFromRaw( _json_scratch_doc, _last_raw_data, _last_raw_len, packet, is_tx, _origin, origin_id, @@ -4127,7 +4063,7 @@ bool MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx, rlen = packet->writeTo(reconstructed); } if (rlen > 0) { - float score = (_radio && !is_tx) ? _radio->packetScore(snr, rlen) : NAN; + float score = captured_score; len = MQTTMessageBuilder::buildPacketJSONFromRaw( _json_scratch_doc, reconstructed, rlen, packet, is_tx, _origin, origin_id, @@ -4326,6 +4262,15 @@ void MQTTBridge::queuePacket(mesh::Packet* packet, bool is_tx) { } } + queued.score = NAN; + if (_radio && !is_tx) { + int score_len = queued.has_raw_data ? queued.raw_len : 0; + if (!score_len && MQTTMessageBuilder::canSerializePacket(packet, sizeof(queued.raw_data))) { + score_len = packet->writeTo(queued.raw_data); + } + if (score_len > 0) queued.score = _radio->packetScore(queued.snr, score_len); + } + // Try to send to queue (non-blocking) if (xQueueSend(_packet_queue_handle, &queued, 0) != pdTRUE) { const MQTTPacketQueuePolicy::EnqueueAction action = @@ -4350,8 +4295,7 @@ void MQTTBridge::queuePacket(mesh::Packet* packet, bool is_tx) { } } - UBaseType_t queue_messages = uxQueueMessagesWaiting(_packet_queue_handle); - _queue_count = queue_messages; + // Queue depth is sampled by the worker; the capture task does not write its cache. #else // Non-ESP32: Use circular buffer const MQTTPacketQueuePolicy::EnqueueAction action = @@ -4391,6 +4335,15 @@ void MQTTBridge::queuePacket(mesh::Packet* packet, bool is_tx) { } } + queued.score = NAN; + if (_radio && !is_tx) { + int score_len = queued.has_raw_data ? queued.raw_len : 0; + if (!score_len && MQTTMessageBuilder::canSerializePacket(packet, sizeof(queued.raw_data))) { + score_len = packet->writeTo(queued.raw_data); + } + if (score_len > 0) queued.score = _radio->packetScore(queued.snr, score_len); + } + _queue_tail = (_queue_tail + 1) % MAX_QUEUE_SIZE; _queue_count++; #endif @@ -4440,10 +4393,6 @@ void MQTTBridge::storeRawRadioData(const uint8_t* raw_data, int len, float snr, // 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, /*attempts_per_server=*/1); -} - // One validated NTP exchange with one named server, on a fresh ephemeral socket. // // Replaces NTPClient, which accepted any datagram that arrived on its fixed @@ -4551,16 +4500,7 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only, int attempts_per return false; } - unsigned long now = millis(); - if (!force && _ntp_synced && (now - _last_ntp_sync) < 5000) { - return false; - } - - static bool sync_in_progress = false; - if (sync_in_progress) { - return false; - } - sync_in_progress = true; + if (!_ntp_schedule.begin((uint32_t)millis(), force)) return false; MQTT_DEBUG_PRINTLN("Syncing time with NTP..."); @@ -4674,8 +4614,8 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only, int attempts_per bool was_ntp_synced = _ntp_synced; _ntp_synced = true; - _last_ntp_sync = millis(); - sync_in_progress = false; + _ntp_schedule.complete((uint32_t)millis(), ntp_server_used + ? NtpSchedule::Result::NetworkTime : NtpSchedule::Result::Holdover); MQTT_DEBUG_PRINTLN("Time synced: %lu (via %s)", epochTime, ntp_server_used ? ntp_server_used : "existing clock"); @@ -4758,41 +4698,23 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only, int attempts_per } MQTT_DEBUG_PRINTLN("NTP sync failed"); - sync_in_progress = false; + _ntp_schedule.complete((uint32_t)millis(), NtpSchedule::Result::Failed); return false; } bool MQTTBridge::requestForcedNtpSync(uint32_t timeout_ms) { + (void)timeout_ms; // Retained for source compatibility; requests never block. if (!isRunning()) 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. - _ntp_force_done = false; - _ntp_force_result = false; - _ntp_force_requested = true; - - // Fire-and-forget: callers on the Arduino loop task (web config batch, and - // the CLI which shares that task) must not block up to 30 s polling the MQTT - // task — that stalls mesh/radio forwarding, portal DNS, and reboot timers. - // The task still performs the sync; the result is observable via - // `get mqtt.ntp.diag`. Blocking callers pass a non-zero timeout. - if (timeout_ms == 0) return true; - - unsigned long start = millis(); - while (!_ntp_force_done) { - if (millis() - start >= timeout_ms) { - MQTT_DEBUG_PRINTLN("Forced NTP sync timed out waiting for MQTT task"); - return false; // task still running; result is ignored when it eventually completes - } - vTaskDelay(pdMS_TO_TICKS(50)); - } - return _ntp_force_result; + publishConfiguration(); + _ntp_force_requested.store(true, std::memory_order_release); + return true; } // Runs on the MQTT task (Core 0). Probes every configured NTP server for connectivity // and records the time each reports. Deliberately does NOT call configTime() or update // the RTC — this is a read-only diagnostic and must leave the system clock untouched. -void MQTTBridge::runNtpDiagProbe() { +void MQTTBridge::runNtpDiagProbe(uint32_t request_id) { + NtpDiagReport report; const char* servers[kMaxNtpServers]; int count = 0; fillNtpServerList(_obs, servers, count); @@ -4803,43 +4725,46 @@ void MQTTBridge::runNtpDiagProbe() { uint32_t epoch = 0; const char* why = nullptr; bool ok = probeNtpServer(servers[i], kNtpMinValidEpoch, &epoch, &why); - NtpDiagResult& r = _ntp_diag_results[i]; + if (_stop_requested.load(std::memory_order_acquire)) { + _ntp_diag_job.cancel(millis()); + return; + } + NtpDiagResult& r = report.servers[i]; strncpy(r.server, servers[i], sizeof(r.server) - 1); r.server[sizeof(r.server) - 1] = '\0'; r.ok = ok; r.epoch = ok ? epoch : 0; - r.why = why; // static literal from NtpValidation/probeNtpServer + snprintf(r.why, sizeof(r.why), "%s", why ? why : "no reply"); } - _ntp_diag_count = count; + report.count = count; + _ntp_diag_job.complete(request_id, millis(), report); } bool MQTTBridge::ntpDiag(char* reply, size_t reply_size, bool verbose) { if (!isRunning() || 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; - _ntp_diag_requested = true; - - unsigned long start = millis(); - while (!_ntp_diag_done) { - if (millis() - start >= 30000) { - snprintf(reply, reply_size, "Error: NTP diag timed out"); - return true; - } - vTaskDelay(pdMS_TO_TICKS(50)); + using Job = ObserverAsyncJob; + auto snapshot = _ntp_diag_job.read(); + if (snapshot.state != Job::State::Complete || (uint32_t)(millis() - snapshot.finished_ms) >= 30000) { + publishConfiguration(); + _ntp_diag_job.request(millis()); + snapshot = _ntp_diag_job.read(); + snprintf(reply, reply_size, "> NTP probe #%lu %s; repeat get mqtt.ntp.diag for result", + (unsigned long)snapshot.id, snapshot.state == Job::State::Running ? "running" : "queued"); + return true; } - + const NtpDiagReport& report = snapshot.result; int ok_count = 0; - for (int i = 0; i < _ntp_diag_count; i++) { - if (_ntp_diag_results[i].ok) ok_count++; + for (int i = 0; i < report.count; i++) { + if (report.servers[i].ok) ok_count++; } if (verbose) { // Detailed table to the serial console; reply carries a short summary (the operator // sees the table above the "-> " line). Mirrors the dumpLogFile() convention. - Serial.printf("NTP diag - %d server(s):\r\n", _ntp_diag_count); - for (int i = 0; i < _ntp_diag_count; i++) { - const NtpDiagResult& r = _ntp_diag_results[i]; + Serial.printf("NTP diag - %d server(s):\r\n", report.count); + for (int i = 0; i < report.count; i++) { + const NtpDiagResult& r = report.servers[i]; if (r.ok) { time_t t = (time_t)r.epoch; struct tm tmv; @@ -4848,16 +4773,20 @@ bool MQTTBridge::ntpDiag(char* reply, size_t reply_size, bool verbose) { r.server, tmv.tm_year + 1900, tmv.tm_mon + 1, tmv.tm_mday, tmv.tm_hour, tmv.tm_min, tmv.tm_sec); } else { - Serial.printf(" %-20s FAIL %s\r\n", r.server, r.why ? r.why : "no reply"); + Serial.printf(" %-20s FAIL %s\r\n", r.server, r.why); } } - snprintf(reply, reply_size, "> NTP diag: %d/%d OK (see console)", ok_count, _ntp_diag_count); + snprintf(reply, reply_size, "> NTP #%lu age:%lus %d/%d OK (see console)", + (unsigned long)snapshot.id, (unsigned long)((uint32_t)(millis() - snapshot.finished_ms) / 1000), + ok_count, report.count); } else { // Compact " ok|fail" list for LoRa, bounded to reply_size. - size_t used = 0; - reply[0] = '\0'; - for (int i = 0; i < _ntp_diag_count; i++) { - const NtpDiagResult& r = _ntp_diag_results[i]; + int header = 0; + replyAppendf(reply, reply_size, &header, "> NTP #%lu age:%lus\n", + (unsigned long)snapshot.id, (unsigned long)((uint32_t)(millis() - snapshot.finished_ms) / 1000)); + size_t used = header; + for (int i = 0; i < report.count; i++) { + const NtpDiagResult& r = report.servers[i]; int n = snprintf(reply + used, reply_size - used, "%s%s %s", used ? "\n" : "", r.server, r.ok ? "ok" : "fail"); if (n < 0 || (size_t)n >= reply_size - used) { @@ -5120,11 +5049,13 @@ void MQTTBridge::logMemoryStatus() { // --------------------------------------------------------------------------- void MQTTBridge::setOrigin(const char* origin) { + if (!mayEditStartupMetadata()) return; strncpy(_origin, origin, sizeof(_origin) - 1); _origin[sizeof(_origin) - 1] = '\0'; } void MQTTBridge::setIATA(const char* iata) { + if (!mayEditStartupMetadata()) return; strncpy(_iata, iata, sizeof(_iata) - 1); _iata[sizeof(_iata) - 1] = '\0'; for (int i = 0; _iata[i]; i++) { @@ -5133,40 +5064,39 @@ void MQTTBridge::setIATA(const char* iata) { } void MQTTBridge::setDeviceID(const char* device_id) { + if (!mayEditStartupMetadata()) return; strncpy(_device_id, device_id, sizeof(_device_id) - 1); _device_id[sizeof(_device_id) - 1] = '\0'; MQTT_DEBUG_PRINTLN("Device ID set to: %s", _device_id); } void MQTTBridge::setFirmwareVersion(const char* firmware_version) { + if (!mayEditStartupMetadata()) return; strncpy(_firmware_version, firmware_version, sizeof(_firmware_version) - 1); _firmware_version[sizeof(_firmware_version) - 1] = '\0'; } void MQTTBridge::setBoardModel(const char* board_model) { + if (!mayEditStartupMetadata()) return; strncpy(_board_model, board_model, sizeof(_board_model) - 1); _board_model[sizeof(_board_model) - 1] = '\0'; } void MQTTBridge::setBuildDate(const char* build_date) { + if (!mayEditStartupMetadata()) return; strncpy(_build_date, build_date, sizeof(_build_date) - 1); _build_date[sizeof(_build_date) - 1] = '\0'; } void MQTTBridge::setMessageTypes(bool status, bool packets, bool raw) { + if (!mayEditStartupMetadata()) return; _status_enabled = status; _packets_enabled = packets; _raw_enabled = raw; } int MQTTBridge::getConnectedBrokers() const { - int count = 0; - for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { - if (_slots[i].enabled && _slots[i].connected) { - count++; - } - } - return count; + return _published.read().connected; } int MQTTBridge::getQueueSize() const { @@ -5182,10 +5112,12 @@ int MQTTBridge::getQueueSize() const { void MQTTBridge::setStatsSources(mesh::Dispatcher* dispatcher, mesh::Radio* radio, mesh::MainBoard* board, mesh::MillisecondClock* ms) { + if (!mayEditStartupMetadata()) return; _dispatcher = dispatcher; _radio = radio; _board = board; _ms = ms; + publishRadioStats(); } #endif diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 1126da0b..6b454d72 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -15,6 +15,10 @@ #include "helpers/AlertFaultPolicy.h" #include "helpers/MQTTEffectiveConfig.h" #include "helpers/MQTTClientState.h" +#include "helpers/NtpSchedule.h" +#include "helpers/MQTTEventChannel.h" +#include "helpers/ObserverConfigMailbox.h" +#include "helpers/ObserverAsyncJob.h" #include #ifdef WITH_SNMP @@ -129,9 +133,10 @@ private: // accept/reject decision for a late callback is made on client_state, which // the bridge task owns. uint32_t generation; + uint32_t incarnation; const MQTTPresetDef* preset; // Points to MQTT_PRESETS[] entry, nullptr for custom/none bool enabled; // true when preset is not "none" - bool connected; // Updated in callbacks + bool connected; // Updated by the worker from event records bool initial_connect_done; // True after first connect() call // JWT auth state (used by preset JWT slots and custom slots with audience set). @@ -192,10 +197,57 @@ private: }; MQTTSlot _slots[RUNTIME_MQTT_SLOTS]; + MqttEventChannel<32> _slot_events; + void processSlotEvents(); + struct SlotRuntimeSnapshot { + char name[32]{}; + char state[16]{}; + bool configured = false; + bool enabled = false; + bool client = false; + bool connected = false; + bool attempted = false; + bool ready = false; + bool circuit_breaker_tripped = false; + ClientState client_state = ClientState::Absent; + uint16_t filter_mask = 0; + unsigned long publish_ok = 0, publish_err = 0; + uint32_t disconnect_count = 0; + unsigned long first_disconnect_time = 0, current_outage_started_ms = 0; + unsigned long last_error_time = 0; + int32_t last_tls_err = 0, last_tls_stack_err = 0; + int last_sock_errno = 0; + uint8_t last_connack_code = 0; + }; + struct RuntimeSnapshot { + uint32_t sampled_ms = 0, sequence = 0; + uint32_t desired_revision = 0, applied_revision = 0; + bool running = false; + int queue = 0, connected = 0, skipped = 0; + uint32_t free_heap = 0, max_heap = 0, event_overflows = 0; + unsigned long outbox = 0, outbox_drops = 0, filtered = 0; + uint8_t neighbors_phase = 0, neighbors_result = 0; + uint32_t neighbors_seconds = 0; + SlotRuntimeSnapshot slots[RUNTIME_MQTT_SLOTS]; + }; + static ObserverMailbox _published; + uint32_t _snapshot_sequence = 0; + void publishRuntimeSnapshot(); + bool mayEditStartupMetadata(); // JWT username shared across all JWT-auth slots (same device identity) char _jwt_username[70]; // Format: v1_{UPPERCASE_PUBLIC_KEY} + struct RadioStatsSnapshot { + int battery_mv = -1, uptime_secs = -1, errors = -1, noise_floor = -999; + int tx_air_secs = -1, rx_air_secs = -1, recv_errors = -1; + int packets_sent = -1, packets_received = -1, state = -1; + uint32_t last_rx_age_ms = 0; + }; + ObserverMailbox _radio_stats; + uint32_t _last_radio_sample_ms = 0; + void publishRadioStats(); // loop task + // Message configuration char _origin[32]; char _iata[8]; @@ -223,6 +275,7 @@ private: bool is_tx; float snr; float rssi; + float score; // Raw radio bytes embedded at enqueue time (Core 1), never shared across cores. // On non-PSRAM boards the queue is smaller (6 items) to offset the per-item cost. uint8_t raw_data[256]; @@ -260,55 +313,27 @@ private: // NTP time sync. The socket is opened per probe and closed again (see // probeNtpServer); nothing listens between syncs. WiFiUDP _ntp_udp; - unsigned long _last_ntp_sync; + NtpSchedule _ntp_schedule; bool _ntp_synced; - bool _ntp_sync_pending; // Flag to trigger NTP sync from loop() instead of event handler bool _slots_setup_done; // Deferred: slots set up after NTP sync int _max_active_slots; // Runtime limit: 5 with PSRAM, 2 without - // Pending slot reconfigure: set from CLI (Core 1), processed by MQTT task (Core 0) - volatile bool _slot_reconfigure_pending[RUNTIME_MQTT_SLOTS]; + // Worker-owned effects requested by applied SDK event records. + bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS]; + bool _status_publish_pending[RUNTIME_MQTT_SLOTS]; - // A broker refusal can invalidate an otherwise clock-valid JWT. The esp-mqtt - // callback sets this and the bridge loop consumes it; byte access is atomic. - volatile bool _slot_force_jwt_mint[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 - // task (Core 0). publishStatusToSlot() touches the shared status doc/buffer/ - // origin that publishStatus() also uses, so it must run only on the bridge - // task — the callback just raises this flag. Same idiom as - // _slot_reconfigure_pending; a single-byte volatile store/load is atomic. - volatile bool _status_publish_pending[RUNTIME_MQTT_SLOTS]; - - // CLI-requested forced NTP sync, marshalled onto the MQTT task (Core 0). - // All NTP I/O must run on Core 0; the CLI thread - // (Core 1) sets _ntp_force_requested and blocks in requestForcedNtpSync() - // until the task publishes the outcome via _ntp_force_result/_ntp_force_done. - // Single-requester assumption: CLI commands are serialized, so at most one - // forced sync is outstanding at a time. - volatile bool _ntp_force_requested; - volatile bool _ntp_force_done; - volatile bool _ntp_force_result; - - // CLI-requested NTP connectivity diagnostic, marshalled onto the MQTT task (Core 0) - // with the same handshake as the forced sync. Probe-only: it queries each server and - // records the reported time but never sets the system clock or the RTC, so the - // system clock is left untouched. Results are written by the task and read by the CLI - // thread once _ntp_diag_done is set. - volatile bool _ntp_diag_requested; - volatile bool _ntp_diag_done; + std::atomic _ntp_force_requested{false}; struct NtpDiagResult { - char server[64]; - bool ok; - uint32_t epoch; // server-reported UTC epoch when ok - // Why a probe failed, as a static literal ("DNS failed", "unsolicited - // reply", ...). Previously every failure looked alike, and a name that never - // resolved could be credited with another server's reply. - const char* why; + char server[64]{}; + bool ok = false; + uint32_t epoch = 0; + char why[48]{}; }; - NtpDiagResult _ntp_diag_results[kMaxNtpServers]; - int _ntp_diag_count; + struct NtpDiagReport { + NtpDiagResult servers[kMaxNtpServers]; + int count = 0; + }; + ObserverAsyncJob _ntp_diag_job; // Cooperative-shutdown handshake. The loop task (Core 1) raises // _stop_requested through the lifecycle Coordinator; the MQTT task (Core 0) @@ -341,6 +366,7 @@ private: uint8_t _staged_raw[LAST_RAW_DATA_SIZE]; int _staged_raw_len = 0; float _staged_snr = 0.0f; + float _last_packet_score = NAN; float _staged_rssi = 0.0f; bool _staged_raw_valid = false; @@ -424,10 +450,8 @@ private: unsigned long _last_memory_check; bool _memory_pressure = false; // Cached max-alloc verdict; re-sampled at most once per interval in publishPacket() so the heap walk isn't paid per-packet under pressure int _skipped_publishes; // Exposed via SNMP; count of publishes skipped when max_alloc is too low - // Packets rejected by the per-slot filters before reaching the queue. Written - // on Core 1 (radio callbacks), read on Core 0 for `mqtt.stats`; a torn read of - // a diagnostic counter is harmless, so no atomic is warranted. - unsigned long _filtered_packets = 0; + // Capture task increments; worker samples for diagnostics. + std::atomic _filtered_packets{0}; // Status publish retry tracking unsigned long _last_status_retry; // Track last retry attempt (separate from successful publish) @@ -436,8 +460,8 @@ private: // Device identity for JWT token creation mesh::LocalIdentity *_identity; - // Cached connection status (updated in callbacks to avoid redundant checks) - bool _cached_has_connected_slots; + // Worker publishes a cheap connection hint to the capture task. + std::atomic _cached_has_connected_slots; // Queue staleness tracking unsigned long _queue_disconnected_since; // 0 = has connected slots @@ -480,10 +504,8 @@ private: // - destroySlotClients() disconnects and deletes each client. Runs once in end(). // - setupSlot() ensures the client exists, then configures it (server, // credentials, CA) and calls connect(). Safe to call again to reconfigure. - // - teardownSlot() only disconnects — it never deletes the client. Leaves - // the mbedTLS/transport state ready for a subsequent setupSlot(). - // This avoids delete/new cycles that shed ~40 KB of mbedTLS buffers per - // reconfigure and fragment the internal heap on non-PSRAM boards. + // - teardownSlot() disconnects without deleting the wrapper. TLS session + // buffers follow the transport connection lifetime, not wrapper lifetime. bool ensureSlotClient(int index); // Allocate this slot's persistent client + callbacks on first use // Stop, destroy and re-allocate this slot's client. Only for configuration // changes that cannot be applied to a live client, and only after the stop is @@ -555,8 +577,8 @@ private: void initializeNetworkInTask(); // Selected-link initialization moved to task #endif bool publishPacket(mesh::Packet* packet, bool is_tx, bool& has_eligible_target, - const uint8_t* raw_data = nullptr, int raw_len = 0, - float snr = 0.0f, float rssi = 0.0f); + const uint8_t* raw_data, int raw_len, + float snr, float rssi, float captured_score); bool publishRaw(mesh::Packet* packet, bool& has_eligible_target); #if defined(WITH_MQTT_NEIGHBORS) // Publishes the pending _neighbors_json_buffer to every connected slot's @@ -566,8 +588,7 @@ private: void queuePacket(mesh::Packet* packet, bool is_tx); void dequeuePacket(); bool isAnySlotConnected(); - void refreshNTP(); // Lightweight periodic NTP refresh (non-blocking) - void runNtpDiagProbe(); // Probe every server for connectivity; never sets the clock. Core 0 only. + void runNtpDiagProbe(uint32_t request_id); // Probe every server for connectivity; never sets the clock. Core 0 only. // One validated NTP exchange with one server on a fresh ephemeral socket. // Core 0 only; never touches the clock. *why receives a static reason literal. bool probeNtpServer(const char* server, uint32_t min_epoch, @@ -613,7 +634,20 @@ private: // Observer config (MQTT/WiFi/timezone/SNMP/alert), persisted to /mqtt.json. // _prefs (held by BridgeBase) still provides upstream fields (freq/sf/node_name…). - MQTTPrefs* _obs = nullptr; + struct RadioMetadata { + char node_name[32]{}; + float freq = 0, bw = 0; + uint8_t sf = 0, cr = 0, disable_fwd = 0; + }; + MQTTPrefs _runtime_prefs{}; + RadioMetadata _radio_metadata{}; + ObserverConfigMailbox _configuration; + uint32_t _applied_revision = 0; + uint32_t _last_config_sample_ms = 0; + const MQTTPrefs* _source_obs = nullptr; // loop-task owned + MQTTPrefs* _obs = &_runtime_prefs; // worker owned + void publishConfiguration(uint32_t reconfigure = 0); // loop task + void applyConfiguration(); // worker NetworkLink* _network = nullptr; public: @@ -729,8 +763,7 @@ public: // Honours a stop acknowledgement that arrived after the deadline: releases the // withheld resources and makes the bridge restartable. Loop task only. void pollLateStopAck(); - // Survives end() clearing the diagnostic singleton, so `get mqtt.status` can - // still explain why a stopped bridge will not come back without a reboot. + // Retains the reason for a stopped bridge until late acknowledgment or restart. static bool stopUnprovenLatched(); static unsigned long getWifiConnectedAtMillis(); @@ -767,6 +800,12 @@ public: unsigned long getSlotCurrentOutageStartMs(int slot_index) const; bool isSlotEnabledAndAttempted(int slot_index) const; const char* getSlotPresetName(int slot_index) const; + struct SlotOutageSnapshot { + bool monitored = false; + uint32_t started_ms = 0; + char name[32]{}; + }; + SlotOutageSnapshot getSlotOutageSnapshot(int slot_index) const; static int getRuntimeSlotCount() { return RUNTIME_MQTT_SLOTS; } /** Max slots that can be connected at once: 5 with PSRAM, 2 without (each * WSS/TLS connection needs ~40KB for mbedTLS buffers). This is the number of @@ -776,35 +815,26 @@ public: /** Resolved origin for MQTT JSON: node_name when mqtt_origin is empty, else mqtt_origin (with quote stripping). */ static void getEffectiveMqttOrigin(const NodePrefs* np, const MQTTPrefs* obs, char* buf, size_t buf_size); static const char* effectiveNtpPrimary(const MQTTPrefs* obs); - /** Sync system clock via NTP. force=true bypasses the 5s post-sync rate limit. - * primary_only=true tests just the effective primary server (no fallback walk) so a - * 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. */ + // Worker only: blocking validated NTP I/O; force bypasses the retry deadline. 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, - * timeout, or if the bridge is not running. */ - bool requestForcedNtpSync(uint32_t timeout_ms = 30000); - /** Probe every configured NTP server (custom primary + built-in fallbacks) for - * connectivity and report each server's reported time WITHOUT touching the system - * clock. Marshals the probe onto the MQTT task (Core 0), then formats on the caller's - * thread. verbose=true prints a detailed table to the serial console and leaves a short - * summary in reply; verbose=false fills reply with a compact " ok|fail" list - * (for LoRa). Returns false if the bridge is not running. */ + // Loop task: queues a primary-server sync and returns immediately. The legacy + // timeout argument is ignored; no production caller may block the mesh loop. + bool requestForcedNtpSync(uint32_t timeout_ms = 0); + // First call queues a probe; repeat to read its ID/age/results. Completed + // results are cached for 30 seconds. Probing never changes the clock. bool ntpDiag(char* reply, size_t reply_size, bool verbose); static void formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPrefs* obs); /** On-demand publish-health + heap snapshot for `get mqtt.stats` (per-slot ok/err, * outbox size, free/max heap, queue depth). */ static void formatMqttStatsReply(char* buf, size_t bufsize); + static void formatRuntimeReply(char* buf, size_t bufsize); /** Structured per-slot snapshot for the webconfig stats endpoint. Same state - * logic as formatMqttStatusReply, same cross-task read semantics. Returns + * logic as formatMqttStatusReply, copied from the published worker snapshot. Returns * false when the bridge is not running, the index is out of range, or the * slot is unconfigured (skip it). */ struct SlotStatusSnapshot { - const char* name; // preset name or "custom" - const char* state; // "inactive" | "wait" | "ok" | "fail" | "disc" + char name[32]; // preset name or "custom" + char state[16]; // "inactive" | "wait" | "ok" | "fail" | "disc" unsigned long publish_ok; unsigned long publish_err; // Raw packet-type allowlist. Kept as the mask rather than canonical text so @@ -812,6 +842,7 @@ public: uint16_t filter_mask; }; static bool getSlotStatusSnapshot(int slot_index, SlotStatusSnapshot* out); + static void appendSlotStatsJson(char* buf, size_t size, int pos); /** True when the selected network is configured and at least one MQTT slot can run. */ static bool isConfigValid(const MQTTPrefs* obs); static void formatSlotDiagReply(char* buf, size_t bufsize, int slot_index); @@ -824,7 +855,7 @@ public: mesh::MainBoard* board, mesh::MillisecondClock* ms); #ifdef WITH_SNMP - void setSNMPAgent(MeshSNMPAgent* agent) { _snmp_agent = agent; } + void setSNMPAgent(MeshSNMPAgent* agent) { if (mayEditStartupMetadata()) _snmp_agent = agent; } #endif };