diff --git a/src/helpers/MQTTTopicRouter.h b/src/helpers/MQTTTopicRouter.h index db7f5cc2..9dfbf17a 100644 --- a/src/helpers/MQTTTopicRouter.h +++ b/src/helpers/MQTTTopicRouter.h @@ -14,6 +14,7 @@ enum MQTTPublicationType { MQTT_PUBLICATION_STATUS = 0, MQTT_PUBLICATION_PACKETS = 1, MQTT_PUBLICATION_RAW = 2, + MQTT_PUBLICATION_NEIGHBORS = 3, }; enum MQTTTopicRouteStyle { @@ -31,6 +32,7 @@ static inline const char* mqttPublicationTypeName(int type) { case MQTT_PUBLICATION_STATUS: return "status"; case MQTT_PUBLICATION_PACKETS: return "packets"; case MQTT_PUBLICATION_RAW: return "raw"; + case MQTT_PUBLICATION_NEIGHBORS: return "neighbors"; default: return NULL; } } @@ -45,7 +47,8 @@ static inline bool mqttWriteTopic(char* buf, size_t buf_size, const char* format } // Build the complete topic for one publication. MeshRank is deliberately -// packets-only; status and raw are unsupported by the current broker contract. +// packets-only; status, raw, and neighbors are unsupported by the current +// broker contract (the type != PACKETS guard below rejects them all). // MeshCore routes require a configured IATA and device id. Custom templates may // omit either placeholder, so their individual values are allowed to be empty. static inline bool mqttBuildPublicationTopic(MQTTTopicRouteStyle style, int type, diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 0a9ccb16..25c67d9b 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -205,6 +205,22 @@ unsigned long MQTTBridge::getWifiConnectedAtMillis() { return s_wifi_connected_at; } +#if defined(WITH_MQTT_NEIGHBORS) +// Compact "time remaining" for the `get mqtt.status` nbr field: "3h12m" / "12m" / "45s". +static void formatDuration(char* buf, size_t len, uint32_t secs) { + if (!buf || len == 0) return; + uint32_t h = secs / 3600; + uint32_t m = (secs % 3600) / 60; + if (h > 0) { + snprintf(buf, len, "%uh%um", (unsigned)h, (unsigned)m); + } else if (m > 0) { + snprintf(buf, len, "%um", (unsigned)m); + } else { + snprintf(buf, len, "%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"; @@ -251,7 +267,33 @@ void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPref } pos += snprintf(buf + pos, bufsize - pos, ", %d: %s (%s)", i + 1, name, state); } - snprintf(buf + pos, bufsize - pos, ", q:%d", q); + // snprintf returns the would-be length, so a full buffer can push pos past + // bufsize; clamp before the remaining appends so bufsize - pos can't underflow. + if (pos >= (int)bufsize) 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. @@ -589,6 +631,17 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg _ntp_diag_done = false; _ntp_diag_count = 0; +#if defined(WITH_MQTT_NEIGHBORS) + // Neighbors publish handoff (buffer allocated in begin() after PSRAM probe). + // std::atomic has no value-initializing default ctor pre-C++20, so set them here. + _neighbors_json_buffer = nullptr; + _neighbors_publish_len = 0; + _neighbors_publish_pending.store(false, std::memory_order_relaxed); + _neighbors_last_result.store(NBR_RESULT_NONE, std::memory_order_relaxed); + _neighbors_phase.store(NBR_SCHEDULED, std::memory_order_relaxed); + _neighbors_secs_until_next.store(0, std::memory_order_relaxed); +#endif + // Initialize JWT username _jwt_username[0] = '\0'; @@ -626,6 +679,13 @@ void MQTTBridge::allocateRuntimeBuffers() { _publish_json_buffer, PUBLISH_JSON_BUFFER_SIZE, psram_malloc)); _status_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( _status_json_buffer, STATUS_JSON_BUFFER_SIZE, psram_malloc)); +#if defined(WITH_MQTT_NEIGHBORS) + // Persistent neighbors JSON buffer. Unlike status/packet there is no stack + // fallback: the feature is PSRAM-gated, so a nullptr simply disables publishing + // (requestPublishNeighbors/publishNeighbors both no-op on nullptr). + _neighbors_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::allocateIfMissing( + _neighbors_json_buffer, NEIGHBORS_JSON_BUFFER_SIZE, psram_malloc)); +#endif MQTT_DEBUG_PRINTLN("Runtime buffers: raw=%s publish=%s status=%s", _last_raw_data ? "PSRAM" : "unavailable", _publish_json_buffer ? "PSRAM" : "stack fallback", @@ -641,6 +701,12 @@ void MQTTBridge::releaseRuntimeBuffers() { _publish_json_buffer, psram_free)); _status_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::release( _status_json_buffer, psram_free)); +#if defined(WITH_MQTT_NEIGHBORS) + _neighbors_json_buffer = static_cast(MQTTRuntimeBufferLifecycle::release( + _neighbors_json_buffer, psram_free)); + _neighbors_publish_len = 0; + _neighbors_publish_pending.store(false, std::memory_order_release); +#endif #endif // Never pair a newly allocated raw buffer with metadata from a prior bridge @@ -1285,6 +1351,25 @@ void MQTTBridge::mqttTaskLoop() { // Process packet queue processPacketQueue(); +#if defined(WITH_MQTT_NEIGHBORS) + // Consume a pending neighbors snapshot handed over by the mesh (Core 1). + // The pending flag stays raised across the whole publish so a second + // request is rejected until this one completes (see requestPublishNeighbors). + if (_neighbors_publish_pending.load(std::memory_order_acquire)) { + bool ok = publishNeighbors(); + _neighbors_last_result.store(ok ? NBR_RESULT_OK : NBR_RESULT_FAIL, + std::memory_order_relaxed); + // MQTT_DEBUG_PRINTLN concatenates its format as a string literal, so the + // argument must be a literal, not a ternary expression. + if (ok) { + MQTT_DEBUG_PRINTLN("Neighbors published"); + } else { + MQTT_DEBUG_PRINTLN("Neighbors publish failed"); + } + _neighbors_publish_pending.store(false, std::memory_order_release); + } +#endif + #ifdef WITH_SNMP // SNMP agent loop — process incoming UDP requests if (_snmp_agent) { @@ -2072,7 +2157,8 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_ static_assert( static_cast(MSG_STATUS) == MQTT_PUBLICATION_STATUS && static_cast(MSG_PACKETS) == MQTT_PUBLICATION_PACKETS && - static_cast(MSG_RAW) == MQTT_PUBLICATION_RAW, + static_cast(MSG_RAW) == MQTT_PUBLICATION_RAW && + static_cast(MSG_NEIGHBORS) == MQTT_PUBLICATION_NEIGHBORS, "topic router enum drift"); if (!mqttTopicSlotIndexValid(index, RUNTIME_MQTT_SLOTS)) return false; @@ -2177,7 +2263,12 @@ void MQTTBridge::publishStatusToSlot(int index) { ); if (len > 0) { - int result = slot.client->publish(status_topic, 1, true, json_buffer, strlen(json_buffer)); + // Honor the preset's retain policy, matching publishStatus() — brokers that + // set allow_retain=false (e.g. waev) reject retained publishes, so this + // on-connect status must not force retain=true. Custom slots default to + // non-retained here too, keeping both status paths consistent. + bool use_retain = slot.preset ? slot.preset->allow_retain : false; + int result = slot.client->publish(status_topic, 1, use_retain, json_buffer, strlen(json_buffer)); if (result <= 0) { MQTT_DEBUG_PRINTLN("MQTT%d status publish failed", index + 1); } @@ -3138,6 +3229,54 @@ bool MQTTBridge::publishRaw(mesh::Packet* packet) { return false; } +#if defined(WITH_MQTT_NEIGHBORS) +// --------------------------------------------------------------------------- +// Periodic neighbors publication +// --------------------------------------------------------------------------- + +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; + // Drop a new snapshot while one is still being published (Core 0 clears the + // flag when done). Acquire pairs with the task loop's release store. + if (_neighbors_publish_pending.load(std::memory_order_acquire)) return; + if (len >= NEIGHBORS_JSON_BUFFER_SIZE) { + len = NEIGHBORS_JSON_BUFFER_SIZE - 1; + } + memcpy(_neighbors_json_buffer, json, len); + _neighbors_json_buffer[len] = '\0'; + _neighbors_publish_len = len; + _neighbors_publish_pending.store(true, std::memory_order_release); +} + +bool MQTTBridge::publishNeighbors() { + if (!_neighbors_json_buffer || _neighbors_publish_len == 0) return false; + if (!_cached_has_connected_slots) return false; + + refreshOriginFromPrefs(); + + bool published = false; + char topic[128]; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + if (_slots[i].enabled && _slots[i].client && _slots[i].connected) { + // MeshRank slots reject non-packets by contract, so buildTopicForSlot + // returns false for them here and the slot is skipped. + if (buildTopicForSlot(i, MSG_NEIGHBORS, topic, sizeof(topic))) { + bool use_retain = _slots[i].preset ? _slots[i].preset->allow_retain : false; + if (publishToSlot(i, topic, _neighbors_json_buffer, use_retain, 1)) { + published = true; + } + } + } + } + return published; +} +#endif // WITH_MQTT_NEIGHBORS + // --------------------------------------------------------------------------- // Queue management // --------------------------------------------------------------------------- diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index a5d405fd..ae2bce44 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -10,6 +10,7 @@ #include "helpers/JWTHelper.h" #include "helpers/MQTTPresets.h" #include "helpers/MQTTLifecycle.h" +#include #ifdef WITH_SNMP class MeshSNMPAgent; // Forward declaration @@ -36,6 +37,14 @@ class MeshSNMPAgent; // Forward declaration #ifdef WITH_MQTT_BRIDGE +// Periodic neighbors publication is PSRAM-only: it needs a persistent ~10 KB JSON +// buffer plus a second transient one while the mesh builds the table, and it keys +// off the mesh neighbor cache (sized by MAX_NEIGHBOURS). Every neighbors-specific +// member, method, and code block in this bridge is gated on WITH_MQTT_NEIGHBORS. +#if defined(BOARD_HAS_PSRAM) && defined(MAX_NEIGHBOURS) && MAX_NEIGHBOURS > 0 +#define WITH_MQTT_NEIGHBORS 1 +#endif + /** * @brief Bridge implementation using MQTT protocol for packet transport * @@ -276,6 +285,26 @@ private: char _status_json_buffer[STATUS_JSON_BUFFER_SIZE]; #endif +#if defined(WITH_MQTT_NEIGHBORS) + // Persistent PSRAM copy of the neighbors-table JSON. The mesh (Core 1) builds + // the payload into its own transient buffer, hands it here via + // requestPublishNeighbors(), and the MQTT task (Core 0) publishes this copy. + // Allocated in allocateRuntimeBuffers()/freed in releaseRuntimeBuffers() like + // the other PSRAM buffers (nullptr if the allocation failed). + char* _neighbors_json_buffer; + size_t _neighbors_publish_len; + // Release/acquire handoff from the mesh loop (Core 1) to the MQTT task (Core 0). + // A second snapshot is dropped while the current one is still publishing. + std::atomic _neighbors_publish_pending; + // Written by the MQTT task (Core 0), read by the CLI (Core 1) for `get mqtt.status`. + 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). Cached schedule + // summary so the wrap-safe millis math stays on the mesh side that owns the timer. + std::atomic _neighbors_phase; + std::atomic _neighbors_secs_until_next; +#endif + // JSON document scratch space — inline StaticJsonDocument keeps the pool off the MQTT // task stack and eliminates two separate heap allocations (fragmentation reduction). StaticJsonDocument _packet_json_doc; @@ -328,7 +357,7 @@ private: mesh::MillisecondClock* _ms; // For uptime // Topic building - enum MQTTMessageType { MSG_STATUS, MSG_PACKETS, MSG_RAW }; + enum MQTTMessageType { MSG_STATUS, MSG_PACKETS, MSG_RAW, MSG_NEIGHBORS }; 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); @@ -370,6 +399,11 @@ private: const uint8_t* raw_data = nullptr, int raw_len = 0, float snr = 0.0f, float rssi = 0.0f); bool publishRaw(mesh::Packet* packet); +#if defined(WITH_MQTT_NEIGHBORS) + // Publishes the pending _neighbors_json_buffer to every connected slot's + // neighbors topic. Runs on the MQTT task (Core 0) only. + bool publishNeighbors(); +#endif void queuePacket(mesh::Packet* packet, bool is_tx); void dequeuePacket(); bool isAnySlotConnected(); @@ -459,6 +493,29 @@ public: void setBuildDate(const char* build_date); void storeRawRadioData(const uint8_t* raw_data, int len, float snr, float rssi); void setMessageTypes(bool status, bool packets, bool raw); + +#if defined(WITH_MQTT_NEIGHBORS) + // Single source of truth for the neighbors JSON size, used by both the bridge's + // persistent buffer and the mesh's transient build buffer. + static const size_t NEIGHBORS_JSON_BUFFER_SIZE = 10240; + + // Called by the mesh (Core 1) once a neighbor-discovery pass has built the + // table JSON. Copies it into the persistent PSRAM buffer and raises the + // publish-pending flag for the MQTT task; a request is dropped if one is + // already in flight or the buffer is unavailable. + void requestPublishNeighbors(const char* json, size_t len); + + // Periodic-neighbors schedule, reported by the mesh loop for `get mqtt.status`. + // The mesh owns the timer; the bridge only caches the summary so the wrap-safe + // millis math stays on the 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 + int getConnectedBrokers() const; int getQueueSize() const; bool isReady() const; diff --git a/test/test_mqtt_topic_router/test_mqtt_topic_router.cpp b/test/test_mqtt_topic_router/test_mqtt_topic_router.cpp index 93e270c7..1156fbd6 100644 --- a/test/test_mqtt_topic_router/test_mqtt_topic_router.cpp +++ b/test/test_mqtt_topic_router/test_mqtt_topic_router.cpp @@ -24,6 +24,7 @@ const TypeCase kTypes[] = { {MQTT_PUBLICATION_STATUS, "status"}, {MQTT_PUBLICATION_PACKETS, "packets"}, {MQTT_PUBLICATION_RAW, "raw"}, + {MQTT_PUBLICATION_NEIGHBORS, "neighbors"}, }; TEST(MQTTTopicRouter, EveryMeshCorePresetSupportsEveryPublicationType) { @@ -57,6 +58,9 @@ TEST(MQTTTopicRouter, MeshRankContractIsPacketsOnly) { EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_RAW, nullptr, IATA, DEVICE, TOKEN, topic, sizeof(topic))); EXPECT_STREQ("", topic); + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_NEIGHBORS, + nullptr, IATA, DEVICE, TOKEN, topic, sizeof(topic))); + EXPECT_STREQ("", topic); } TEST(MQTTTopicRouter, MeshCoreRequiresUsableIataAndDevice) { @@ -178,9 +182,11 @@ TEST(MQTTTopicRouter, PublicationTypeEnumValuesAreFrozen) { EXPECT_EQ(0, MQTT_PUBLICATION_STATUS); EXPECT_EQ(1, MQTT_PUBLICATION_PACKETS); EXPECT_EQ(2, MQTT_PUBLICATION_RAW); + EXPECT_EQ(3, MQTT_PUBLICATION_NEIGHBORS); EXPECT_STREQ("status", mqttPublicationTypeName(MQTT_PUBLICATION_STATUS)); EXPECT_STREQ("packets", mqttPublicationTypeName(MQTT_PUBLICATION_PACKETS)); EXPECT_STREQ("raw", mqttPublicationTypeName(MQTT_PUBLICATION_RAW)); + EXPECT_STREQ("neighbors", mqttPublicationTypeName(MQTT_PUBLICATION_NEIGHBORS)); } } // namespace