From b3bd5ecd3c6f18a81855460604e1c795c7a501da Mon Sep 17 00:00:00 2001 From: agessaman Date: Fri, 17 Jul 2026 07:59:42 -0700 Subject: [PATCH] feat(mqtt): send all message types to MeshRank and harden neighbors flow MeshRank slots now publish every message type (status/packets/raw/neighbors) under meshrank/uplink/{token}/{device}/, matching the standard MeshCore topic layout, instead of packets only. The four duplicated topic-suffix ternaries are folded into messageTypeSuffix(). Fix the discover.neighbors -> discover.scopes race: a scopes request issued while a zero-hop discovery is still collecting responses (from discover.neighbors or the periodic timer) now queues behind that 60-second window and runs against the refreshed table, reporting the wait in its reply. A queued one-shot request is no longer cancelled by `mqtt.neighbors off`; only the periodic timer's own refresh is. Precondition checks are shared via neighborDiscoverReady() so queuing cannot report OK for a request that will fail once the window closes. Add the periodic-neighbors schedule to `get mqtt.status` as a trailing nbr:/ field (time to next publish + last result). The mesh owns the timer and reports a summary to the bridge across the core boundary via atomics. Also clamp the running offset in formatMqttStatusReply: snprintf returns the untruncated length, so a full 6-slot line could push the offset past the buffer and underflow the remaining size on the next append. Reachable today with real preset-name combinations at the 160-byte reply size. --- MQTT_IMPLEMENTATION.md | 14 ++-- docs/cli_commands.md | 11 ++- examples/simple_repeater/MyMesh.cpp | 81 ++++++++++++++++++---- examples/simple_repeater/MyMesh.h | 2 + src/helpers/MQTTPresets.h | 2 +- src/helpers/bridges/MQTTBridge.cpp | 101 ++++++++++++++++++++++------ src/helpers/bridges/MQTTBridge.h | 17 +++++ 7 files changed, 185 insertions(+), 43 deletions(-) diff --git a/MQTT_IMPLEMENTATION.md b/MQTT_IMPLEMENTATION.md index 188fc4f4..60ec72f7 100644 --- a/MQTT_IMPLEMENTATION.md +++ b/MQTT_IMPLEMENTATION.md @@ -296,7 +296,7 @@ set mqtt3.preset meshrank set mqtt3.token FE1B34242C5938C39225310081FD6718 ``` -The token is generated on the MeshRank website and is tied to your account. MeshRank only receives packet data (no status or raw messages). +The token is generated on the MeshRank website and is tied to your account. MeshRank receives the same message types as any other preset (`status`, `packets`, `raw`, `neighbors`), each under `meshrank/uplink/{token}/{device_id}/`, subject to the usual `mqtt.status`/`mqtt.raw`/`mqtt.neighbors` toggles. #### Example: Configure MeshMapper on Slot 3 ```bash @@ -376,7 +376,7 @@ These settings apply across all MQTT slots: - `get mqtt.iata` - Get IATA code - `get mqtt.presets` - List available MQTT presets (paginated, comma-separated) - `get mqtt.presets ` - Continue list from index shown in `... next:` -- `get mqtt.status` - Get MQTT status summary (connection info per slot) +- `get mqtt.status` - Get MQTT status summary (connection info per slot, plus the periodic neighbors schedule when `mqtt.neighbors` is on) - `get mqtt.packets` - Get packet message setting (on/off) - `get mqtt.raw` - Get raw message setting (on/off) - `get mqtt.rx` - Get RX packet uplinking setting (on/off) @@ -509,11 +509,15 @@ Full packet data with RF characteristics and metadata. Minimal raw packet data for map integration. ### Neighbors Topic: `meshcore/{IATA}/{DEVICE_PUBLIC_KEY}/neighbors` -Cached zero-hop repeater neighbors with SNR, last-heard age, and flood-allowed scopes (retained, QoS 1). Published on `discover.scopes` or periodically when `mqtt.neighbors` is enabled (PSRAM observer builds only). +Cached zero-hop repeater neighbors with SNR, last-heard age, and flood-allowed scopes. Published on `discover.scopes` or periodically when `mqtt.neighbors` is enabled (PSRAM observer builds only). Goes to every configured slot's `neighbors` topic at QoS 1, retained only where the preset allows it. -Periodic publishing first runs a 60-second zero-hop neighbor refresh equivalent to `discover.neighbors`, then queries the refreshed table for scopes and publishes when the scope-query phase completes. Manual `discover.scopes` remains a one-shot query of the current cache. +Periodic publishing first runs a 60-second zero-hop neighbor refresh equivalent to `discover.neighbors`, then queries the refreshed table for scopes and publishes when the scope-query phase completes. -**Note**: `{DEVICE_PUBLIC_KEY}` is the device's public key in hexadecimal format (64 characters). +Manual `discover.scopes` normally queries the current cache in one shot. If a `discover.neighbors` refresh is already collecting responses — whether started from the CLI or by the periodic timer — the scope queries are queued behind its 60-second window instead, so they run against the refreshed table. The reply reports the wait, e.g. `OK - scopes queued (47s discovery remaining)`. A queued one-shot request survives `set mqtt.neighbors off`; only the periodic timer's own refresh is cancelled by it. + +While `mqtt.neighbors` is on, `get mqtt.status` appends `nbr: /` — time to the next automatic publish (`3h12m`, `12m`, `45s`, or `active`/`due`) and the last publish result (`ok`, `failed`, or `none`). + +**Note**: `{DEVICE_PUBLIC_KEY}` is the device's public key in hexadecimal format (64 characters). MeshRank slots use `meshrank/uplink/{token}/{DEVICE_PUBLIC_KEY}/neighbors` instead. ## JSON Message Formats diff --git a/docs/cli_commands.md b/docs/cli_commands.md index d8618e67..55f8cd64 100644 --- a/docs/cli_commands.md +++ b/docs/cli_commands.md @@ -118,7 +118,9 @@ This document provides an overview of CLI commands that can be sent to MeshCore **Usage:** - `discover.scopes` -Queries cached zero-hop repeater neighbors for flood-allowed region scopes (`ANON_REQ_TYPE_REGIONS`), then publishes a retained JSON snapshot to `meshcore/{IATA}/{device}/neighbors`. Includes this node's own scopes in the `self` object. +Queries cached zero-hop repeater neighbors for flood-allowed region scopes (`ANON_REQ_TYPE_REGIONS`), then publishes a JSON snapshot to each configured MQTT slot's `neighbors` topic. Includes this node's own scopes in the `self` object. + +If a zero-hop neighbor discovery is already collecting responses — from `discover.neighbors` or from the periodic `mqtt.neighbors` refresh — the scope queries are queued until its 60-second response window ends, so they run against the refreshed table rather than the stale cache. The reply reports the wait, e.g. `OK - scopes queued (47s discovery remaining)`. A queued request is not cancelled by `set mqtt.neighbors off`. **Note:** Requires `WITH_MQTT_BRIDGE`, `MAX_NEIGHBOURS`, and PSRAM (`BOARD_HAS_PSRAM`). Returns `Err - not supported (requires PSRAM)` on other builds. @@ -1125,10 +1127,15 @@ Requires WiFi connected and the MQTT bridge running. **Parameters:** - `hours`: Interval between automatic `discover.scopes` runs (**12-336**, default **24**) -**Note:** Requires PSRAM. Publishes to `meshcore/{IATA}/{device}/neighbors` (retained, QoS 1). Use `discover.scopes` for a one-shot publish. +**Note:** Requires PSRAM. Publishes to each configured MQTT slot's `neighbors` topic at QoS 1 (retained only when that preset allows it). Use `discover.scopes` for a one-shot publish. Automatic publishes run in two stages: first the repeater performs the same 60-second zero-hop refresh as `discover.neighbors`, then it queries the refreshed neighbors for scopes and publishes after those queries complete. +While enabled, `get mqtt.status` reports the schedule as a trailing `nbr: /` field, e.g. `nbr: 3h12m/ok`: + +- `` — time until the next automatic publish (`3h12m`, `12m`, `45s`), or `active` while a refresh or scope query is in flight, or `due` when a publish is waiting on the bridge or WiFi to come up. +- `` — result of the most recent publish attempt: `ok`, `failed`, or `none` if none has been attempted since boot. + --- #### View or change the speed of the bridge (RS-232 only) diff --git a/examples/simple_repeater/MyMesh.cpp b/examples/simple_repeater/MyMesh.cpp index c6010871..098d60b1 100644 --- a/examples/simple_repeater/MyMesh.cpp +++ b/examples/simple_repeater/MyMesh.cpp @@ -1036,6 +1036,7 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc neighbor_discover_count = 0; neighbor_discover_active = false; neighbor_table_refresh_active = false; + neighbor_table_refresh_periodic = false; neighbor_discover_until = 0; next_neighbors_publish = 0; self_scopes_buf[0] = 0; @@ -1451,6 +1452,25 @@ void MyMesh::handleCommand(uint32_t sender_timestamp, char *command, char *reply while (*sub == ' ') sub++; if (*sub != 0) { strcpy(reply, "Err - discover.scopes has no options"); + } else if (pending_discover_tag != 0 && + !millisHasNowPassed(pending_discover_until) && + !neighbor_discover_active) { + // Reuse an in-flight zero-hop discovery rather than querying the old + // neighbor table. An explicit command takes ownership of a periodic + // refresh so `mqtt.neighbors off` cannot cancel this one-shot request. + // Validate up front so queuing cannot report OK for a request that is + // certain to fail once the discovery window closes. + if (!neighborDiscoverReady(reply)) { + // reply already set + } else { + neighbor_table_refresh_active = true; + neighbor_table_refresh_periodic = false; + long remaining_ms = (long)(pending_discover_until - futureMillis(0)); + unsigned remaining_secs = remaining_ms > 0 + ? (unsigned)(((unsigned long)remaining_ms + 999UL) / 1000UL) : 0; + sprintf(reply, "OK - scopes queued (%us discovery remaining)", remaining_secs); + MESH_DEBUG_PRINTLN("Neighbor scopes queued behind active discovery (%us remaining)", remaining_secs); + } } else if (!startNeighborDiscover(reply)) { // reply already set } @@ -1535,25 +1555,33 @@ void MyMesh::loop() { if (neighbor_discover_active) { loopNeighborDiscover(); } else if (neighbor_table_refresh_active) { - if (!periodic_neighbors_enabled) { + if (neighbor_table_refresh_periodic && !periodic_neighbors_enabled) { // Turning periodic publishing off cancels a refresh that has not yet // reached the scope-query/publish phase. Re-enabling starts fresh. + // pending_discover_tag is left alone: this refresh may have joined a + // discover.neighbors the user started, and clearing the tag would drop + // its remaining responses. An unused tag expires on its own. neighbor_table_refresh_active = false; - pending_discover_tag = 0; + neighbor_table_refresh_periodic = false; next_neighbors_publish = 0; } else if (pending_discover_tag == 0 || millisHasNowPassed(pending_discover_until)) { // The zero-hop discover.neighbors response window is complete. Freeze the // refreshed table into the scope-query overlay, then publish when those // per-neighbor requests finish. + bool was_periodic = neighbor_table_refresh_periodic; pending_discover_tag = 0; neighbor_table_refresh_active = false; + neighbor_table_refresh_periodic = false; char tmp_reply[80]; + const char* origin_str = was_periodic ? "periodic" : "manual"; if (startNeighborDiscover(tmp_reply)) { - MESH_DEBUG_PRINTLN("MQTT periodic %s", tmp_reply); + MESH_DEBUG_PRINTLN("MQTT %s %s", origin_str, tmp_reply); } else { - // Avoid retrying a failed refresh on every loop iteration. - next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval); - MESH_DEBUG_PRINTLN("MQTT periodic neighbor scope discovery failed: %s", tmp_reply); + if (periodic_neighbors_enabled) { + // Avoid retrying a failed refresh on every loop iteration. + next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval); + } + MESH_DEBUG_PRINTLN("MQTT %s neighbor scope discovery failed: %s", origin_str, tmp_reply); } } } else if (periodic_neighbors_enabled && bridge && bridge->isRunning()) { @@ -1562,9 +1590,29 @@ void MyMesh::loop() { // Refresh the zero-hop neighbor cache first (the same operation as the // discover.neighbors CLI command). Scope queries begin after its 60-second // collection window instead of publishing a stale cached table. - sendNodeDiscoverReq(); + if (pending_discover_tag == 0 || millisHasNowPassed(pending_discover_until)) { + pending_discover_tag = 0; + sendNodeDiscoverReq(); + MESH_DEBUG_PRINTLN("MQTT periodic neighbor table refresh started"); + } else { + MESH_DEBUG_PRINTLN("MQTT periodic refresh joined active neighbor discovery"); + } neighbor_table_refresh_active = true; - MESH_DEBUG_PRINTLN("MQTT periodic neighbor table refresh started"); + neighbor_table_refresh_periodic = true; + } + } + + // Report the periodic-neighbors schedule for `get mqtt.status`. The timer lives + // here, so the wrap-safe millis math stays on this side of the handoff. + if (bridge) { + if (neighbor_discover_active || neighbor_table_refresh_active) { + bridge->setNeighborsSchedule(MQTTBridge::NBR_ACTIVE, 0); + } else if (next_neighbors_publish == 0 || millisHasNowPassed(next_neighbors_publish)) { + bridge->setNeighborsSchedule(MQTTBridge::NBR_DUE, 0); + } else { + long remaining_ms = (long)(next_neighbors_publish - futureMillis(0)); + uint32_t remaining_secs = remaining_ms > 0 ? (uint32_t)(remaining_ms / 1000) : 0; + bridge->setNeighborsSchedule(MQTTBridge::NBR_SCHEDULED, remaining_secs); } } #endif @@ -1755,11 +1803,7 @@ void MyMesh::loopNeighborDiscover() { finishNeighborDiscover(); } -bool MyMesh::startNeighborDiscover(char* reply) { - if (neighbor_discover_active) { - strcpy(reply, "Err - neighbor discover already active"); - return false; - } +bool MyMesh::neighborDiscoverReady(char* reply) { #if defined(ESP_PLATFORM) if (!psramFound()) { strcpy(reply, "Err - PSRAM not available"); @@ -1770,6 +1814,17 @@ bool MyMesh::startNeighborDiscover(char* reply) { strcpy(reply, "Err - MQTT bridge not running"); return false; } + return true; +} + +bool MyMesh::startNeighborDiscover(char* reply) { + if (neighbor_discover_active) { + strcpy(reply, "Err - neighbor discover already active"); + return false; + } + if (!neighborDiscoverReady(reply)) { + return false; // reply already set + } getLocalScopes(self_scopes_buf, sizeof(self_scopes_buf)); neighbor_discover_count = 0; diff --git a/examples/simple_repeater/MyMesh.h b/examples/simple_repeater/MyMesh.h index bb2c6698..b223a62f 100644 --- a/examples/simple_repeater/MyMesh.h +++ b/examples/simple_repeater/MyMesh.h @@ -153,11 +153,13 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks { uint8_t neighbor_discover_count; bool neighbor_discover_active; bool neighbor_table_refresh_active; + bool neighbor_table_refresh_periodic; unsigned long neighbor_discover_until; unsigned long next_neighbors_publish; char self_scopes_buf[96]; bool sendAnonRegionsReq(const mesh::Identity& target, uint32_t& tag); + bool neighborDiscoverReady(char* reply); bool startNeighborDiscover(char* reply); void loopNeighborDiscover(); void finishNeighborDiscover(); diff --git a/src/helpers/MQTTPresets.h b/src/helpers/MQTTPresets.h index 8ab063e0..38289d72 100644 --- a/src/helpers/MQTTPresets.h +++ b/src/helpers/MQTTPresets.h @@ -23,7 +23,7 @@ enum MQTTAuthType : uint8_t { enum MQTTTopicStyle : uint8_t { MQTT_TOPIC_MESHCORE, // meshcore/{iata}/{device_id}/{status|packets|raw} - MQTT_TOPIC_MESHRANK, // meshrank/uplink/{token}/{device_id}/packets (packets only) + MQTT_TOPIC_MESHRANK, // meshrank/uplink/{token}/{device_id}/{status|packets|raw|neighbors} }; struct MQTTPresetDef { diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 72a5bf26..93959837 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -201,6 +201,19 @@ unsigned long MQTTBridge::getWifiConnectedAtMillis() { return s_wifi_connected_at; } +#if defined(WITH_MQTT_NEIGHBORS) +// Compact "3h12m" / "12m" / "45s" for CLI replies that share a 160-byte buffer. +static void formatDuration(char* buf, size_t bufsize, uint32_t secs) { + if (secs >= 3600) { + snprintf(buf, bufsize, "%uh%um", (unsigned)(secs / 3600), (unsigned)((secs % 3600) / 60)); + } else if (secs >= 60) { + snprintf(buf, bufsize, "%um", (unsigned)(secs / 60)); + } else { + snprintf(buf, bufsize, "%us", (unsigned)secs); + } +} +#endif + 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"; @@ -246,8 +259,40 @@ void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPref state = "disc"; } pos += snprintf(buf + pos, bufsize - pos, ", %d: %s (%s)", i + 1, name, state); + // snprintf returns the untruncated length, so clamp before reusing pos as an + // offset: an overflowing pos would walk past buf and underflow bufsize - pos. + if (pos >= (int)bufsize) { pos = (int)bufsize - 1; break; } } - snprintf(buf + pos, bufsize - pos, ", q:%d", q); + if (pos < (int)bufsize - 1) { + pos += snprintf(buf + pos, bufsize - pos, ", q:%d", q); + if (pos >= (int)bufsize) pos = (int)bufsize - 1; + } + +#if defined(WITH_MQTT_NEIGHBORS) + // Periodic neighbors: time to next publish + how the last one went. + if (obs && obs->mqtt_neighbors_enabled && pos < (int)bufsize - 1) { + char when[16]; + switch (b->_neighbors_phase.load(std::memory_order_relaxed)) { + 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)); + break; + } + const char* last; + switch (b->_neighbors_last_result.load(std::memory_order_relaxed)) { + case NBR_RESULT_OK: last = "ok"; break; + case NBR_RESULT_FAIL: last = "failed"; break; + default: last = "none"; break; + } + snprintf(buf + pos, bufsize - pos, ", nbr: %s/%s", when, last); + } +#endif } // On-demand publish-health + heap snapshot for the `get mqtt.stats` CLI command. @@ -533,6 +578,9 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg _neighbors_json_buffer = (char*)psram_malloc(NEIGHBORS_JSON_BUFFER_SIZE); _neighbors_publish_pending = false; _neighbors_publish_len = 0; + _neighbors_last_result = NBR_RESULT_NONE; + _neighbors_phase = NBR_SCHEDULED; + _neighbors_secs_until_next = 0; #endif #else memset(_last_raw_data, 0, sizeof(_last_raw_data)); @@ -1059,7 +1107,10 @@ void MQTTBridge::mqttTaskLoop() { #if defined(WITH_MQTT_NEIGHBORS) if (_neighbors_publish_pending.load(std::memory_order_acquire)) { - if (publishNeighbors()) { + bool ok = publishNeighbors(); + _neighbors_last_result.store(ok ? NBR_RESULT_OK : NBR_RESULT_FAIL, + std::memory_order_relaxed); + if (ok) { MQTT_DEBUG_PRINTLN("Neighbors published"); } else { MQTT_DEBUG_PRINTLN("Neighbors publish failed"); @@ -1864,11 +1915,17 @@ bool MQTTBridge::publishToAllSlots(const char* topic, const char* payload, bool // Topic building - resolves the correct topic for a given slot and message type. // Presets use hardcoded topic logic; custom slots support user-defined templates. // --------------------------------------------------------------------------- +const char* MQTTBridge::messageTypeSuffix(MQTTMessageType type) { + switch (type) { + case MSG_STATUS: return "status"; + case MSG_PACKETS: return "packets"; + case MSG_RAW: return "raw"; + default: return "neighbors"; + } +} + bool MQTTBridge::substituteTopicTemplate(const char* tmpl, MQTTMessageType type, int slot_index, char* buf, size_t buf_size) { - const char* type_str = (type == MSG_STATUS) ? "status" - : (type == MSG_PACKETS) ? "packets" - : (type == MSG_RAW) ? "raw" - : "neighbors"; + const char* type_str = messageTypeSuffix(type); const char* token = _obs->mqtt_slot_token[slot_index]; size_t out = 0; @@ -1917,20 +1974,18 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_ // Preset slots: use hardcoded topic logic if (slot.preset) { if (slot.preset->topic_style == MQTT_TOPIC_MESHRANK) { - // MeshRank: packets only, uses per-slot token in topic path - if (type != MSG_PACKETS) return false; + // MeshRank uses a per-slot token in its topic path. Publish every message + // type using the same suffixes as the standard MeshCore topic layout. const char* token = _obs->mqtt_slot_token[index]; if (!token || token[0] == '\0') return false; - snprintf(topic_buf, buf_size, "meshrank/uplink/%s/%s/packets", token, _device_id); + snprintf(topic_buf, buf_size, "meshrank/uplink/%s/%s/%s", token, _device_id, + messageTypeSuffix(type)); return true; } // MQTT_TOPIC_MESHCORE (default for all other presets) if (!isIATAValid()) return false; - const char* type_str = (type == MSG_STATUS) ? "status" - : (type == MSG_PACKETS) ? "packets" - : (type == MSG_RAW) ? "raw" - : "neighbors"; - snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id, type_str); + snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id, + messageTypeSuffix(type)); return true; } @@ -1940,11 +1995,8 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_ } // Default: meshcore format if (!isIATAValid()) return false; - const char* type_str = (type == MSG_STATUS) ? "status" - : (type == MSG_PACKETS) ? "packets" - : (type == MSG_RAW) ? "raw" - : "neighbors"; - snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id, type_str); + snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id, + messageTypeSuffix(type)); return true; } @@ -1958,7 +2010,7 @@ void MQTTBridge::publishStatusToSlot(int index) { // Build per-slot topic (handles IATA check for meshcore, token check for meshrank) char status_topic[128]; if (!buildTopicForSlot(index, MSG_STATUS, status_topic, sizeof(status_topic))) { - return; // Slot doesn't support status (e.g., meshrank) or missing required config + return; // Slot is missing required topic configuration } // Reuse pre-allocated buffer to avoid heap alloc/free churn under memory pressure. @@ -2755,8 +2807,8 @@ bool MQTTBridge::publishStatus() { } } } - // If no connected slot accepts status topics (e.g. meshrank is packets-only), - // treat as success to avoid infinite retry loops + // If no connected slot accepts status topics, treat as success to avoid + // infinite retry loops. if (published || !any_slot_wants_status) { if (published) MQTT_DEBUG_PRINTLN("Status published"); return true; @@ -2767,6 +2819,11 @@ bool MQTTBridge::publishStatus() { } #if defined(WITH_MQTT_NEIGHBORS) +void MQTTBridge::setNeighborsSchedule(NeighborsPhase phase, uint32_t secs_until_next) { + _neighbors_phase.store((uint8_t)phase, std::memory_order_relaxed); + _neighbors_secs_until_next.store(secs_until_next, std::memory_order_relaxed); +} + void MQTTBridge::requestPublishNeighbors(const char* json, size_t len) { if (!_neighbors_json_buffer || !json || len == 0) return; if (_neighbors_publish_pending.load(std::memory_order_acquire)) return; diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index 23dd452a..5e7e2d01 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -268,6 +268,12 @@ private: // Release/acquire handoff from the Arduino loop (Core 1) to the MQTT task // (Core 0). A second snapshot is dropped while the current one is publishing. std::atomic _neighbors_publish_pending; + // Written by the MQTT task (Core 0), read by the CLI (Core 1). + enum NeighborsResult : uint8_t { NBR_RESULT_NONE, NBR_RESULT_OK, NBR_RESULT_FAIL }; + std::atomic _neighbors_last_result; + // Written by the mesh loop (Core 1), read by the CLI (Core 1). + std::atomic _neighbors_phase; + std::atomic _neighbors_secs_until_next; #endif #if defined(BOARD_HAS_PSRAM) char* _publish_json_buffer; @@ -331,6 +337,7 @@ private: // Topic building enum MQTTMessageType { MSG_STATUS, MSG_PACKETS, MSG_RAW, MSG_NEIGHBORS }; + static const char* messageTypeSuffix(MQTTMessageType type); bool buildTopicForSlot(int index, MQTTMessageType type, char* topic_buf, size_t buf_size); bool substituteTopicTemplate(const char* tmpl, MQTTMessageType type, int slot_index, char* buf, size_t buf_size); @@ -495,6 +502,16 @@ public: #if defined(WITH_MQTT_NEIGHBORS) void requestPublishNeighbors(const char* json, size_t len); static const size_t NEIGHBORS_JSON_BUFFER_SIZE = 10240; + + // Periodic-neighbors schedule, reported by the mesh loop for `get mqtt.status`. + // The mesh owns the timer; the bridge only caches the summary, which keeps the + // wrap-safe millis math on the one side that already has those helpers. + enum NeighborsPhase : uint8_t { + NBR_SCHEDULED, // waiting for the next publish; secs_until_next is valid + NBR_ACTIVE, // zero-hop refresh or scope queries in flight + NBR_DUE, // publish is due, waiting on the bridge/WiFi to come up + }; + void setNeighborsSchedule(NeighborsPhase phase, uint32_t secs_until_next); #endif static const char* wifiReasonStr(uint8_t reason); static const char* tlsErrorStr(int32_t err);