From 8d7a47abf7510c62d8e2472daf69f6d30ace83d9 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 19 Jul 2026 20:41:20 -0700 Subject: [PATCH 1/7] feat(mqtt): add neighbors prefs fields (flex-compatible v1 layout) Append mqtt_neighbors_enabled(u8) + mqtt_neighbors_interval(u32) to the observer tail of MQTTPrefs. The layout is kept byte-identical to the flex neighbors build: the enable flag lands in the old struct's zeroed trailing padding (offset 2857) and the interval begins exactly at the former baseline (2860), so sizeof grows 2860 -> 2864 (net +4 bytes). offsetof static_asserts lock the layout so a mismatch fails the build. The codec now accepts three v1 payload sizes: register 2860 as a "pre-neighbors" Current payload so an in-lineage upgrade reads its existing /mqtt_prefs and defaults the neighbors tail (off / 24h). Because 2864 is the shared Current baseline, a /mqtt_prefs written by either the flex build or this firmware is interchangeable. Add the 12/24/336h interval constants, neighbors defaults, and a load-time interval clamp that keeps persisted values inside the signed-delta millis() scheduling window. Extend the host codec suite with a pre-neighbors migration case and neighbors round-trip coverage. --- src/helpers/CommonCLI.cpp | 10 +++++ src/helpers/MQTTDefaults.h | 5 +++ src/helpers/MQTTPrefsCodec.h | 7 ++++ src/helpers/MQTTPrefsStorage.h | 36 ++++++++++++++++- .../test_mqtt_prefs_codec.cpp | 40 +++++++++++++++++++ 5 files changed, 96 insertions(+), 2 deletions(-) diff --git a/src/helpers/CommonCLI.cpp b/src/helpers/CommonCLI.cpp index 02569e4c..b430374b 100644 --- a/src/helpers/CommonCLI.cpp +++ b/src/helpers/CommonCLI.cpp @@ -807,6 +807,16 @@ void CommonCLI::loadMQTTPrefs( mqtt_rewrite_pending = true; MESH_DEBUG_PRINTLN("MQTT: Migrated observer settings from legacy /com_prefs trailing block"); } + + // Keep persisted values inside the signed-delta millis() scheduling window. + // This also repairs any manually-written or experimental value from firmware + // that briefly accepted intervals longer than the supported two-week cap. + if (_mqtt_prefs.mqtt_neighbors_interval < MQTT_NEIGHBORS_MIN_INTERVAL_MS || + _mqtt_prefs.mqtt_neighbors_interval > MQTT_NEIGHBORS_MAX_INTERVAL_MS) { + _mqtt_prefs.mqtt_neighbors_interval = MQTT_NEIGHBORS_DEFAULT_INTERVAL_MS; + MESH_DEBUG_PRINTLN("MQTT: invalid neighbors interval reset to %u hours", + (unsigned)MQTT_NEIGHBORS_DEFAULT_INTERVAL_HOURS); + } _legacy_tail.valid = false; if (mqtt_rewrite_pending) { diff --git a/src/helpers/MQTTDefaults.h b/src/helpers/MQTTDefaults.h index f3c117eb..6b4c9ef1 100644 --- a/src/helpers/MQTTDefaults.h +++ b/src/helpers/MQTTDefaults.h @@ -100,6 +100,11 @@ static inline void applyMQTTDefaults(MQTTPrefs* prefs) { prefs->alert_wifi_minutes = 30; prefs->alert_mqtt_minutes = 240; prefs->alert_min_interval_min = 60; + + // Neighbors publishing defaults off; a defaulted tail is a valid 24h interval + // (not 0) so an in-lineage upgrade from a pre-neighbors payload is sane. + prefs->mqtt_neighbors_enabled = 0; + prefs->mqtt_neighbors_interval = MQTT_NEIGHBORS_DEFAULT_INTERVAL_MS; } #endif // WITH_MQTT_BRIDGE diff --git a/src/helpers/MQTTPrefsCodec.h b/src/helpers/MQTTPrefsCodec.h index 85629dff..715d33f2 100644 --- a/src/helpers/MQTTPrefsCodec.h +++ b/src/helpers/MQTTPrefsCodec.h @@ -36,6 +36,7 @@ struct DecodePlan { }; static const size_t kV1PreObserverPayloadSize = MQTT_PREFS_V1_PRE_OBSERVER_PAYLOAD_SIZE; +static const size_t kV1PreNeighborsPayloadSize = MQTT_PREFS_V1_PRE_NEIGHBORS_PAYLOAD_SIZE; static const size_t kV1BaselinePayloadSize = MQTT_PREFS_V1_FULL_PAYLOAD_SIZE; static const size_t kEncodedSize = sizeof(MQTTPrefsHeader) + kV1BaselinePayloadSize; @@ -99,6 +100,12 @@ inline DecodePlan classify(const uint8_t* prefix, size_t prefix_read, size_t fil if (header.payload_len == kV1BaselinePayloadSize) { return {Source::Current, false, false, true, kV1BaselinePayloadSize}; } + if (header.payload_len == kV1PreNeighborsPayloadSize) { + // Written by observer/webconfig firmware before the neighbors tail + // existed. The observer fields ARE present; only the neighbors tail is + // missing, so it loads and keeps its defaults (off / 24h). + return {Source::Current, false, false, true, kV1PreNeighborsPayloadSize}; + } if (header.payload_len == kV1PreObserverPayloadSize) { return {Source::Current, false, false, false, kV1PreObserverPayloadSize}; } diff --git a/src/helpers/MQTTPrefsStorage.h b/src/helpers/MQTTPrefsStorage.h index fd898694..ffa90c58 100644 --- a/src/helpers/MQTTPrefsStorage.h +++ b/src/helpers/MQTTPrefsStorage.h @@ -108,12 +108,35 @@ struct MQTTPrefs { uint16_t alert_min_interval_min; char alert_hashtag[24]; char alert_region[31]; + + // Neighbors publishing (PSRAM boards only). Appended at the end of the + // observer tail so a shorter (pre-neighbors) /mqtt_prefs payload from earlier + // firmware still loads with these defaulting off/24h; keeps the format at + // VERSION 1. Field order and sizes are kept byte-identical to the flex + // neighbors build so a /mqtt_prefs written by either firmware is + // interchangeable (see the offsetof static_asserts below). + uint8_t mqtt_neighbors_enabled; + uint32_t mqtt_neighbors_interval; }; -// Version-1 has exactly two layouts this firmware can decode. Never infer a +// Neighbor discovery is scheduled with the wrap-safe millis() helpers, whose +// signed-delta comparison requires intervals below INT32_MAX ms. The 336h +// (two-week) cap stays comfortably inside that range. +static const uint32_t MQTT_NEIGHBORS_MIN_INTERVAL_HOURS = 12; +static const uint32_t MQTT_NEIGHBORS_MAX_INTERVAL_HOURS = 336; +static const uint32_t MQTT_NEIGHBORS_DEFAULT_INTERVAL_HOURS = 24; +static const uint32_t MQTT_NEIGHBORS_MIN_INTERVAL_MS = MQTT_NEIGHBORS_MIN_INTERVAL_HOURS * 3600000UL; +static const uint32_t MQTT_NEIGHBORS_MAX_INTERVAL_MS = MQTT_NEIGHBORS_MAX_INTERVAL_HOURS * 3600000UL; +static const uint32_t MQTT_NEIGHBORS_DEFAULT_INTERVAL_MS = MQTT_NEIGHBORS_DEFAULT_INTERVAL_HOURS * 3600000UL; + +// Version-1 has three payload layouts this firmware can decode. Never infer a // compatible payload from an arbitrary shorter size: raw prefs have no checksum. +// - PRE_OBSERVER (2736): stops before the observer tail (snmp_*/alert_*). +// - PRE_NEIGHBORS (2860): full observer tail, no neighbors fields yet. +// - FULL (2864): current baseline, with the neighbors tail. static const size_t MQTT_PREFS_V1_PRE_OBSERVER_PAYLOAD_SIZE = 2736; -static const size_t MQTT_PREFS_V1_FULL_PAYLOAD_SIZE = 2860; +static const size_t MQTT_PREFS_V1_PRE_NEIGHBORS_PAYLOAD_SIZE = 2860; +static const size_t MQTT_PREFS_V1_FULL_PAYLOAD_SIZE = 2864; // /mqtt_prefs starts with a self-describing 8-byte header. Headerless files // are deployed legacy layouts and continue to be distinguished by size. @@ -234,6 +257,15 @@ static_assert(offsetof(MQTTPrefs, snmp_enabled) == MQTT_PREFS_V1_PRE_OBSERVER_PA "v1 pre-observer /mqtt_prefs boundary changed"); static_assert(sizeof(MQTTPrefs) == MQTT_PREFS_V1_FULL_PAYLOAD_SIZE, "v1 /mqtt_prefs payload layout changed"); +// Lock the neighbors tail to the flex neighbors build's layout so a /mqtt_prefs +// written by either firmware is byte-for-byte interchangeable. The enable flag +// lands in the old struct's zeroed trailing padding (offset 2857), and the +// interval begins exactly at the pre-neighbors payload size (2860) so a +// pre-neighbors read stops right before it and the interval keeps its default. +static_assert(offsetof(MQTTPrefs, mqtt_neighbors_enabled) == 2857, + "neighbors enable flag must sit at the flex-compatible offset"); +static_assert(offsetof(MQTTPrefs, mqtt_neighbors_interval) == MQTT_PREFS_V1_PRE_NEIGHBORS_PAYLOAD_SIZE, + "neighbors interval offset must equal the pre-neighbors payload size"); static_assert(sizeof(OldMQTTPrefs) == 472, "frozen pre-slot /mqtt_prefs layout changed"); static_assert(sizeof(PreWifiPowerOldMQTTPrefs) == 472, "frozen pre-WiFi-power /mqtt_prefs layout changed"); static_assert(offsetof(OldMQTTPrefs, wifi_power_save) == 144, diff --git a/test/test_mqtt_prefs_codec/test_mqtt_prefs_codec.cpp b/test/test_mqtt_prefs_codec/test_mqtt_prefs_codec.cpp index 2909fa5f..6d1fba4b 100644 --- a/test/test_mqtt_prefs_codec/test_mqtt_prefs_codec.cpp +++ b/test/test_mqtt_prefs_codec/test_mqtt_prefs_codec.cpp @@ -226,6 +226,8 @@ TEST(MQTTPrefsCodec, CurrentVersionedPayloadRoundTripsExactly) { strncpy(source.mqtt_origin, "current-node", sizeof(source.mqtt_origin) - 1); strncpy(source.mqtt_slot_password[2], "preserve-me", sizeof(source.mqtt_slot_password[2]) - 1); strncpy(source.alert_region, "PNW", sizeof(source.alert_region) - 1); + source.mqtt_neighbors_enabled = 1; + source.mqtt_neighbors_interval = MQTT_NEIGHBORS_MAX_INTERVAL_MS; std::vector bytes(Codec::kEncodedSize); ASSERT_EQ(Codec::kEncodedSize, Codec::encode(source, bytes.data(), bytes.size())); @@ -266,6 +268,44 @@ TEST(MQTTPrefsCodec, CompatibleShortV1PayloadPreservesDefaultsBeyondObserverBoun EXPECT_EQ(5, loaded.radio_watchdog_minutes); } +TEST(MQTTPrefsCodec, PreNeighborsV1PayloadLoadsObserverFieldsAndDefaultsNeighborsTail) { + // A /mqtt_prefs written by observer/webconfig firmware before the neighbors + // tail existed: full observer fields, 2860-byte v1 payload. It must still load + // as Current (observer fields present) with the neighbors tail defaulted. + MQTTPrefs source = defaults(); + strncpy(source.mqtt_origin, "pre-neighbors-node", sizeof(source.mqtt_origin) - 1); + strncpy(source.alert_region, "PNW", sizeof(source.alert_region) - 1); + source.snmp_enabled = 1; + source.alert_enabled = 1; + source.mqtt_neighbors_enabled = 0; // old struct's byte 2857 was zero padding + source.mqtt_neighbors_interval = 0x11223344; // must NOT survive a 2860-byte read + + std::vector bytes(sizeof(MQTTPrefsHeader) + Codec::kV1PreNeighborsPayloadSize, 0); + writeHeader(&bytes, MQTT_PREFS_VERSION, + static_cast(Codec::kV1PreNeighborsPayloadSize)); + memcpy(bytes.data() + sizeof(MQTTPrefsHeader), &source, Codec::kV1PreNeighborsPayloadSize); + + const Codec::DecodePlan plan = classify(bytes); + ASSERT_EQ(Codec::Source::Current, plan.source); + ASSERT_EQ(Codec::kV1PreNeighborsPayloadSize, plan.payload_len); + ASSERT_FALSE(plan.preserve_file); + ASSERT_TRUE(plan.observer_fields_present); + + MQTTPrefs loaded = defaults(); + loaded.mqtt_neighbors_enabled = 1; // pretend stale + loaded.mqtt_neighbors_interval = MQTT_NEIGHBORS_DEFAULT_INTERVAL_MS; // caller's defaulted tail + memcpy(&loaded, bytes.data() + sizeof(MQTTPrefsHeader), plan.payload_len); + + EXPECT_STREQ("pre-neighbors-node", loaded.mqtt_origin); + EXPECT_STREQ("PNW", loaded.alert_region); + EXPECT_EQ(1, loaded.snmp_enabled); + EXPECT_EQ(1, loaded.alert_enabled); + // Enable flag sits at offset 2857 (inside the 2860 read) -> takes the file's 0. + // Interval begins at 2860 (beyond the read) -> keeps the caller's default. + EXPECT_EQ(0u, loaded.mqtt_neighbors_enabled); + EXPECT_EQ(MQTT_NEIGHBORS_DEFAULT_INTERVAL_MS, loaded.mqtt_neighbors_interval); +} + TEST(MQTTPrefsCodec, CorruptOrShortVersionedInputsArePreserved) { Codec::DecodePlan plan = Codec::classify(nullptr, 0, 0); EXPECT_EQ(Codec::Source::Corrupt, plan.source); From e36aee04d408bcdd75702aa2cde28e0443383d09 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 19 Jul 2026 22:05:16 -0700 Subject: [PATCH 2/7] feat(mqtt): add neighbors JSON payload builder (host-tested) Add buildNeighborsMessage to the pure MQTTPayloadBuilder core and a thin delegating wrapper + NeighborsMessageEntry alias on MQTTMessageBuilder, so the neighbors topic is built by the same firmware-facing API as status/ packet/raw while the layout logic stays exercisable by native tests. The document is bounded to the publish buffer: entries arrive ordered most- to least-useful and the tail is dropped once the next entry would overflow, so a fixed PSRAM buffer can never be handed truncated JSON. Uses ArduinoJson v7 idioms (.to()/.add()) to stay warning-clean under -Werror, unlike the deprecated createNested* forms. Adds three test_mqtt_payload_builder cases: self+entry round-trip, empty table / null scopes, and bounded-growth tail-drop under a tight buffer. --- src/helpers/MQTTMessageBuilder.cpp | 16 ++++ src/helpers/MQTTMessageBuilder.h | 17 ++++ src/helpers/MQTTPayloadBuilder.cpp | 45 +++++++++++ src/helpers/MQTTPayloadBuilder.h | 23 ++++++ .../test_mqtt_payload_builder.cpp | 81 +++++++++++++++++++ 5 files changed, 182 insertions(+) diff --git a/src/helpers/MQTTMessageBuilder.cpp b/src/helpers/MQTTMessageBuilder.cpp index fb0d4399..2fe44d6b 100644 --- a/src/helpers/MQTTMessageBuilder.cpp +++ b/src/helpers/MQTTMessageBuilder.cpp @@ -102,6 +102,22 @@ int MQTTMessageBuilder::buildRawMessage( origin, origin_id, timestamp, raw, buffer, buffer_size); } +int MQTTMessageBuilder::buildNeighborsMessage( + JsonDocument& doc, + const char* origin, + const char* origin_id, + const char* timestamp, + const char* self_scopes, + const NeighborsMessageEntry* neighbors, + int neighbor_count, + char* buffer, + size_t buffer_size +) { + return MQTTPayloadBuilder::buildNeighborsMessage( + doc, origin, origin_id, timestamp, self_scopes, neighbors, neighbor_count, + buffer, buffer_size); +} + int MQTTMessageBuilder::buildPacketJSON( JsonDocument& doc, mesh::Packet* packet, diff --git a/src/helpers/MQTTMessageBuilder.h b/src/helpers/MQTTMessageBuilder.h index 8e3a6d45..fc0c4705 100644 --- a/src/helpers/MQTTMessageBuilder.h +++ b/src/helpers/MQTTMessageBuilder.h @@ -2,6 +2,7 @@ #include "MeshCore.h" #include +#include "MQTTPayloadBuilder.h" #include #include @@ -153,6 +154,22 @@ public: size_t buffer_size ); + // Neighbors table entry + JSON builder. The layout logic lives in the pure, + // host-tested MQTTPayloadBuilder; this is the firmware-facing alias/delegate, + // matching the status/packet/raw builders. + using NeighborsMessageEntry = MQTTPayloadBuilder::NeighborsMessageEntry; + static int buildNeighborsMessage( + JsonDocument& doc, + const char* origin, + const char* origin_id, + const char* timestamp, + const char* self_scopes, + const NeighborsMessageEntry* neighbors, + int neighbor_count, + char* buffer, + size_t buffer_size + ); + /** * Convert packet to JSON message * diff --git a/src/helpers/MQTTPayloadBuilder.cpp b/src/helpers/MQTTPayloadBuilder.cpp index 260fac5f..5a40e502 100644 --- a/src/helpers/MQTTPayloadBuilder.cpp +++ b/src/helpers/MQTTPayloadBuilder.cpp @@ -184,3 +184,48 @@ int MQTTPayloadBuilder::buildRawMessage( return serializeComplete(root, buffer, buffer_size); } + +int MQTTPayloadBuilder::buildNeighborsMessage( + JsonDocument& doc, + const char* origin, + const char* origin_id, + const char* timestamp, + const char* self_scopes, + const NeighborsMessageEntry* neighbors, + int neighbor_count, + char* buffer, + size_t buffer_size +) { + if (!buffer || buffer_size == 0) return 0; + + doc.clear(); + JsonObject root = doc.to(); + root["timestamp"] = timestamp; + root["origin"] = origin; + root["origin_id"] = origin_id; + + JsonObject self = root["self"].to(); + self["scopes"] = self_scopes ? self_scopes : ""; + + JsonArray arr = root["neighbors"].to(); + if (measureJson(root) >= buffer_size) return 0; + + for (int i = 0; i < neighbor_count; i++) { + JsonObject nb = arr.add(); + nb["pubkey"] = neighbors[i].pubkey_hex; + nb["snr"] = neighbors[i].snr; + nb["heard_secs_ago"] = neighbors[i].heard_secs_ago; + nb["scopes"] = neighbors[i].scopes ? neighbors[i].scopes : ""; + nb["status"] = neighbors[i].status; + + // Entries arrive ordered most- to least-useful. Stop as soon as the next + // one would fill the fixed publish buffer, dropping the remaining tail so + // document growth stays bounded. + if (measureJson(root) >= buffer_size) { + arr.remove(arr.size() - 1); + break; + } + } + + return serializeComplete(root, buffer, buffer_size); +} diff --git a/src/helpers/MQTTPayloadBuilder.h b/src/helpers/MQTTPayloadBuilder.h index 57793b13..a6fa333d 100644 --- a/src/helpers/MQTTPayloadBuilder.h +++ b/src/helpers/MQTTPayloadBuilder.h @@ -68,5 +68,28 @@ public: char* buffer, size_t buffer_size ); + + struct NeighborsMessageEntry { + const char* pubkey_hex; + float snr; + uint32_t heard_secs_ago; + const char* scopes; + const char* status; + }; + + // Build neighbors-table JSON for the meshcore/{iata}/{device}/neighbors topic. + // Callers order entries most- to least-useful; document growth is bounded to + // buffer_size and the remaining tail is dropped once the next entry won't fit. + static int buildNeighborsMessage( + JsonDocument& doc, + const char* origin, + const char* origin_id, + const char* timestamp, + const char* self_scopes, + const NeighborsMessageEntry* neighbors, + int neighbor_count, + char* buffer, + size_t buffer_size + ); }; diff --git a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp index 7dfcf965..d7f81130 100644 --- a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp +++ b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp @@ -227,6 +227,87 @@ TEST(MQTTPayloadBuilder, MaximumRepresentativePacketAndRawPayloadsRemainValid) { EXPECT_EQ(510U, strlen(parsed_raw["data"].as())); } +TEST(MQTTPayloadBuilder, NeighborsMessageRoundTripsSelfAndEntries) { + MQTTPayloadBuilder::NeighborsMessageEntry neighbors[] = { + {"0011223344556677", 9.75f, 42, "DEN,APRS", "active"}, + {"8899AABBCCDDEEFF", -3.5f, 3600, "", "stale"}, + }; + + JsonDocument scratch; + char buffer[1024]; + int len = MQTTPayloadBuilder::buildNeighborsMessage( + scratch, "DEN Repeater", "0123456789ABCDEF", kTimestamp, "DEN,APRS", + neighbors, 2, buffer, sizeof(buffer)); + + ASSERT_GT(len, 0); + EXPECT_EQ(static_cast(len), strlen(buffer)); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + EXPECT_STREQ("2026-07-18T12:34:56.123456+00:00", parsed["timestamp"].as()); + EXPECT_STREQ("DEN Repeater", parsed["origin"].as()); + EXPECT_STREQ("0123456789ABCDEF", parsed["origin_id"].as()); + EXPECT_STREQ("DEN,APRS", parsed["self"]["scopes"].as()); + + JsonArray arr = parsed["neighbors"].as(); + ASSERT_EQ(2U, arr.size()); + EXPECT_STREQ("0011223344556677", arr[0]["pubkey"].as()); + EXPECT_FLOAT_EQ(9.75f, arr[0]["snr"].as()); + EXPECT_EQ(42U, arr[0]["heard_secs_ago"].as()); + EXPECT_STREQ("DEN,APRS", arr[0]["scopes"].as()); + EXPECT_STREQ("active", arr[0]["status"].as()); + EXPECT_STREQ("8899AABBCCDDEEFF", arr[1]["pubkey"].as()); + EXPECT_STREQ("", arr[1]["scopes"].as()); + EXPECT_STREQ("stale", arr[1]["status"].as()); +} + +TEST(MQTTPayloadBuilder, NeighborsMessageHandlesEmptyTableAndNullScopes) { + JsonDocument scratch; + char buffer[256]; + int len = MQTTPayloadBuilder::buildNeighborsMessage( + scratch, "node", "id", kTimestamp, nullptr, nullptr, 0, + buffer, sizeof(buffer)); + + ASSERT_GT(len, 0); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + EXPECT_STREQ("", parsed["self"]["scopes"].as()); + JsonArray arr = parsed["neighbors"].as(); + ASSERT_TRUE(arr.isNull() == false); + EXPECT_EQ(0U, arr.size()); +} + +TEST(MQTTPayloadBuilder, NeighborsMessageDropsTailWhenBufferFills) { + // Twenty entries far exceed a tight buffer; the builder must emit a prefix + // that still parses as complete JSON rather than truncating mid-document. + MQTTPayloadBuilder::NeighborsMessageEntry neighbors[20]; + static char keys[20][17]; + for (int i = 0; i < 20; i++) { + snprintf(keys[i], sizeof(keys[i]), "%016X", i); + neighbors[i].pubkey_hex = keys[i]; + neighbors[i].snr = static_cast(i); + neighbors[i].heard_secs_ago = static_cast(i) * 10U; + neighbors[i].scopes = "DEN"; + neighbors[i].status = "active"; + } + + JsonDocument scratch; + char buffer[512]; + int len = MQTTPayloadBuilder::buildNeighborsMessage( + scratch, "node", "id", kTimestamp, "DEN", neighbors, 20, + buffer, sizeof(buffer)); + + ASSERT_GT(len, 0); + EXPECT_LT(static_cast(len), sizeof(buffer)); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + JsonArray arr = parsed["neighbors"].as(); + ASSERT_FALSE(arr.isNull()); + EXPECT_GT(arr.size(), 0U); + EXPECT_LT(arr.size(), 20U); + // Kept entries are the head of the input, in order. + EXPECT_STREQ(keys[0], arr[0]["pubkey"].as()); +} + } // namespace int main(int argc, char** argv) { From de320bc4df3ddd3a9d2602e2d0bf5bdd2ff1c3c3 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 19 Jul 2026 22:20:21 -0700 Subject: [PATCH 3/7] feat(mqtt): add neighbors publish path to MQTTBridge Port the periodic-neighbors publication from mqtt-bridge-implementation-flex, adapted to this branch's structure: - Add MQTT_PUBLICATION_NEIGHBORS ("neighbors") to the pure MQTTTopicRouter instead of flex's messageTypeSuffix() helper (this branch already routes every publication type through mqttBuildPublicationTopic()). Neighbors is a MeshCore/custom publication type, so it resolves to meshcore/{iata}/{device}/neighbors and honors custom templates. - Deliberately do NOT port flex's "all message types to MeshRank" change: this branch documents and host-tests a packets-only MeshRank contract (MQTTPresets.h, MQTT_IMPLEMENTATION.md, MeshRankContractIsPacketsOnly). So neighbors follows status/raw and is rejected on MeshRank slots. - WITH_MQTT_NEIGHBORS guard (PSRAM + MAX_NEIGHBOURS) gates all new surface. - MSG_NEIGHBORS message type + enum-drift static_assert. - Persistent ~10KB PSRAM neighbors buffer allocated/freed via the existing MQTTRuntimeBufferLifecycle path (allocate/release), not the ctor as flex did. - Core1->Core0 handoff: requestPublishNeighbors() (mesh) fills the buffer with a release store; the MQTT task consumes it with an acquire load, publishes via publishNeighbors() (QoS1, retain = preset->allow_retain, custom=false), and clears the pending flag. A second snapshot is dropped while one is in flight. - setNeighborsSchedule()/NeighborsPhase let the mesh report the timer summary; formatMqttStatusReply() gains a "nbr: /" field via formatDuration. Also fix the on-connect status publish (publishStatusToSlot) to honor preset->allow_retain instead of hardcoding retain=true, matching the periodic publishStatus() path. Brokers with allow_retain=false (e.g. the waev MeshCore preset) reject retained publishes, so the on-connect status was being dropped there. This is flex followup 028a5dca, reconciled to this branch's custom-slot default of non-retained. Extends the host topic-router test to cover the neighbors type across all routes and freezes the new enum value. Bridge itself is on-target only. --- src/helpers/MQTTTopicRouter.h | 5 +- src/helpers/bridges/MQTTBridge.cpp | 145 +++++++++++++++++- src/helpers/bridges/MQTTBridge.h | 59 ++++++- .../test_mqtt_topic_router.cpp | 6 + 4 files changed, 210 insertions(+), 5 deletions(-) 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 From 34a822011122f0a9b625b76cab57f45d1c307a78 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 19 Jul 2026 22:38:00 -0700 Subject: [PATCH 4/7] feat(mqtt): add neighbor-scope discovery + periodic publish to MyMesh Port the neighbor-discovery state machine from mqtt-bridge-implementation-flex, adapted to this branch's MyMesh (WebConfig members shifted the insertion points; applied by content). - Two-stage periodic refresh in loop() driven by mqtt_neighbors_interval: stage 1 is a zero-hop sendNodeDiscoverReq() (reuses the existing 60s window), stage 2 (startNeighborDiscover) fires one anon-regions scope query per heard neighbour, then finishNeighborDiscover() builds the table JSON via MQTTMessageBuilder::buildNeighborsMessage and hands it to bridge->requestPublishNeighbors(). - Peer overlay at NEIGHBOR_DISCOVER_PEER_BASE lets scope-query RESPONSE packets from non-ACL neighbours decrypt: searchPeersByHash prepends heard neighbours (bounded by MAX_CLIENTS), getPeerSharedSecret derives the secret on the fly, and onPeerDataRecv routes both overlay-index and ACL-client-that-is-a-neighbour responses into handleNeighborDiscoverResponse. - Entries ordered most- to least-useful (recent, then stronger SNR) so the JSON builder's tail-drop keeps the useful head. - `discover.scopes` CLI command (manual trigger), with a WITH_MQTT_BRIDGE stub replying "requires PSRAM" on non-PSRAM builds. - Reports schedule to the bridge each loop via setNeighborsSchedule(). - Uses ArduinoJson v7 JsonDocument (not deprecated DynamicJsonDocument). All gated on WITH_MQTT_NEIGHBORS. Reuses the existing MQTTBridge* member. Verified: T_Beam_S3_Supreme_SX1262_repeater_observer_mqtt builds [SUCCESS]. --- examples/simple_repeater/MyMesh.cpp | 366 +++++++++++++++++++++++++++- examples/simple_repeater/MyMesh.h | 38 +++ 2 files changed, 403 insertions(+), 1 deletion(-) diff --git a/examples/simple_repeater/MyMesh.cpp b/examples/simple_repeater/MyMesh.cpp index febcd5d5..1e3201d1 100644 --- a/examples/simple_repeater/MyMesh.cpp +++ b/examples/simple_repeater/MyMesh.cpp @@ -666,7 +666,22 @@ void MyMesh::onAnonDataRecv(mesh::Packet *packet, const uint8_t *secret, const m int MyMesh::searchPeersByHash(const uint8_t *hash) { int n = 0; - for (int i = 0; i < acl.getNumClients(); i++) { +#if defined(WITH_MQTT_NEIGHBORS) + // While a neighbor-scope discovery is active, overlay the heard neighbours + // that are NOT already ACL clients so their RESPONSE packets can be decoded. + // Overlay indices are offset by NEIGHBOR_DISCOVER_PEER_BASE to keep them + // distinct from real ACL indices. + if (neighbor_discover_active) { + for (int i = 0; i < neighbor_discover_count && n < MAX_CLIENTS; i++) { + auto& nb = neighbours[neighbor_discover[i].neighbour_idx]; + if (acl.getClient(nb.id.pub_key, PUB_KEY_SIZE) != nullptr) continue; + if (nb.heard_timestamp > 0 && nb.id.isHashMatch(hash)) { + matching_peer_indexes[n++] = NEIGHBOR_DISCOVER_PEER_BASE + i; + } + } + } +#endif + for (int i = 0; i < acl.getNumClients() && n < MAX_CLIENTS; i++) { if (acl.getClientByIdx(i)->id.isHashMatch(hash)) { matching_peer_indexes[n++] = i; // store the INDEXES of matching contacts (for subsequent 'peer' methods) } @@ -676,6 +691,16 @@ int MyMesh::searchPeersByHash(const uint8_t *hash) { void MyMesh::getPeerSharedSecret(uint8_t *dest_secret, int peer_idx) { int i = matching_peer_indexes[peer_idx]; +#if defined(WITH_MQTT_NEIGHBORS) + // Overlay entries have no precomputed shared secret; derive it on the fly. + if (neighbor_discover_active && i >= NEIGHBOR_DISCOVER_PEER_BASE) { + int oi = i - NEIGHBOR_DISCOVER_PEER_BASE; + if (oi >= 0 && oi < neighbor_discover_count) { + self_id.calcSharedSecret(dest_secret, neighbours[neighbor_discover[oi].neighbour_idx].id); + return; + } + } +#endif if (i >= 0 && i < acl.getNumClients()) { // lookup pre-calculated shared_secret memcpy(dest_secret, acl.getClientByIdx(i)->shared_secret, PUB_KEY_SIZE); @@ -707,11 +732,34 @@ void MyMesh::onAdvertRecv(mesh::Packet *packet, const mesh::Identity &id, uint32 void MyMesh::onPeerDataRecv(mesh::Packet *packet, uint8_t type, int sender_idx, const uint8_t *secret, uint8_t *data, size_t len) { int i = matching_peer_indexes[sender_idx]; +#if defined(WITH_MQTT_NEIGHBORS) + // Overlay response: a heard neighbour (not an ACL client) answering our + // anon-regions scope query. Consume it and stop — it is not a client packet. + if (neighbor_discover_active && i >= NEIGHBOR_DISCOVER_PEER_BASE) { + int oi = i - NEIGHBOR_DISCOVER_PEER_BASE; + if (type == PAYLOAD_TYPE_RESPONSE && oi >= 0 && oi < neighbor_discover_count) { + handleNeighborDiscoverResponse(oi, data, len); + } + return; + } +#endif if (i < 0 || i >= acl.getNumClients()) { // get from our known_clients table (sender SHOULD already be known in this context) MESH_DEBUG_PRINTLN("onPeerDataRecv: invalid peer idx: %d", i); return; } ClientInfo* client = acl.getClientByIdx(i); +#if defined(WITH_MQTT_NEIGHBORS) + // A neighbour that IS an ACL client resolves to a normal index above, so a + // scope-query response from it lands here — match it against the overlay. + if (neighbor_discover_active && type == PAYLOAD_TYPE_RESPONSE) { + for (int oi = 0; oi < neighbor_discover_count; oi++) { + auto& nb = neighbours[neighbor_discover[oi].neighbour_idx]; + if (client->id.matches(nb.id) && handleNeighborDiscoverResponse(oi, data, len)) { + return; + } + } + } +#endif if (type == PAYLOAD_TYPE_REQ) { // request (from a Known admin client!) uint32_t timestamp; @@ -986,6 +1034,16 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc pending_discover_tag = 0; pending_discover_until = 0; +#if defined(WITH_MQTT_NEIGHBORS) + 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; +#endif + memset(default_scope.key, 0, sizeof(default_scope.key)); } @@ -1519,6 +1577,36 @@ void MyMesh::handleCommand(uint32_t sender_timestamp, char *command, char *reply sendNodeDiscoverReq(); strcpy(reply, "OK - Discover sent"); } +#if defined(WITH_MQTT_NEIGHBORS) + } else if (memcmp(command, "discover.scopes", 15) == 0) { + const char* sub = command + 15; + 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) { + // A zero-hop table refresh is already collecting; queue the scope pass + // behind it (as a manual, non-periodic request) rather than starting a + // second refresh. + if (!neighborDiscoverReady(reply)) { + // reply already set by neighborDiscoverReady + } 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 by startNeighborDiscover + } +#elif defined(WITH_MQTT_BRIDGE) + } else if (memcmp(command, "discover.scopes", 15) == 0) { + strcpy(reply, "Err - not supported (requires PSRAM)"); +#endif } else{ _cli.handleCommand(sender_timestamp, command, reply); // common CLI commands } @@ -1615,6 +1703,66 @@ void MyMesh::loop() { _alerter.onLoop(now); #endif +#if defined(WITH_MQTT_NEIGHBORS) + // Two-stage periodic neighbors publication: + // stage 1 - zero-hop node-discover refreshes the neighbour table (60s window) + // stage 2 - anon-regions scope query per neighbour (startNeighborDiscover) + // then the table JSON is published and the next cycle is rescheduled. + bool periodic_neighbors_enabled = _cli.getObserverPrefs()->mqtt_neighbors_enabled; + if (neighbor_discover_active) { + loopNeighborDiscover(); + } else if (neighbor_table_refresh_active) { + if (neighbor_table_refresh_periodic && !periodic_neighbors_enabled) { + // periodic switched off mid-refresh -> cancel (leave pending_discover_tag alone) + neighbor_table_refresh_active = false; + neighbor_table_refresh_periodic = false; + next_neighbors_publish = 0; + } else if (pending_discover_tag == 0 || millisHasNowPassed(pending_discover_until)) { + // 60s zero-hop window done -> begin the per-neighbour scope queries + 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 %s %s", origin_str, tmp_reply); + } else { + if (periodic_neighbors_enabled) { + 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()) { + if (next_neighbors_publish == 0 || + (next_neighbors_publish != 0 && millisHasNowPassed(next_neighbors_publish))) { + 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; + neighbor_table_refresh_periodic = true; + } + } + + // Report the schedule state back to the bridge for `get mqtt.status`. + 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 + #ifdef WITH_SNMP // Push radio stats to SNMP agent every 2 seconds if (_snmp_agent.isRunning()) { @@ -1635,6 +1783,222 @@ void MyMesh::loop() { #endif } +#if defined(WITH_MQTT_NEIGHBORS) +#include "helpers/MQTTMessageBuilder.h" +#if defined(ESP_PLATFORM) +#include +#endif + +// This node's own non-flood scope names, same source the anon-regions server +// reply uses. Empty string when the node has no scoped regions. +void MyMesh::getLocalScopes(char* buf, size_t len) { + if (!buf || len == 0) return; + buf[0] = 0; + region_map.exportNamesTo(buf, (int)len, REGION_DENY_FLOOD); +} + +// Client side of the anon-regions request (the server side is handleAnonRegionsReq). +// Inner payload: {tag(4)}{ANON_REQ_TYPE_REGIONS}{0x00 = zero-hop reply path}. +bool MyMesh::sendAnonRegionsReq(const mesh::Identity& target, uint32_t& tag) { + uint8_t secret[PUB_KEY_SIZE]; + self_id.calcSharedSecret(secret, target); + + tag = getRTCClock()->getCurrentTimeUnique(); + uint8_t inner[6]; + memcpy(inner, &tag, 4); + inner[4] = ANON_REQ_TYPE_REGIONS; + inner[5] = 0x00; // request a zero-hop reply path + + mesh::Packet* pkt = createAnonDatagram(PAYLOAD_TYPE_ANON_REQ, self_id, target, secret, inner, sizeof(inner)); + if (!pkt) return false; + sendDirect(pkt, NULL, 0, 0); + return true; +} + +// Match a RESPONSE against the pending overlay entry by tag; copy its scope +// string (payload after the 8-byte {tag}{clock} header) into the entry. +bool MyMesh::handleNeighborDiscoverResponse(int overlay_idx, const uint8_t* data, size_t len) { + if (overlay_idx < 0 || overlay_idx >= neighbor_discover_count) return false; + NeighborDiscoverEntry& entry = neighbor_discover[overlay_idx]; + if (entry.status != ND_PENDING || len < 8) return false; + + uint32_t tag; + memcpy(&tag, data, 4); + if (tag != entry.tag) return false; + + size_t scope_len = len - 8; + if (scope_len >= sizeof(entry.scopes)) { + scope_len = sizeof(entry.scopes) - 1; + } + memcpy(entry.scopes, &data[8], scope_len); + entry.scopes[scope_len] = 0; + entry.status = ND_RESPONDED; + return true; +} + +// Publish-ordering: most recently heard first, then stronger SNR, then pubkey. +// The JSON builder drops the tail if the buffer fills, so the head must be the +// most useful entries. +static bool neighborPublishEntryComesBefore( + const MQTTMessageBuilder::NeighborsMessageEntry& lhs, + const MQTTMessageBuilder::NeighborsMessageEntry& rhs) { + if (lhs.heard_secs_ago != rhs.heard_secs_ago) { + return lhs.heard_secs_ago < rhs.heard_secs_ago; // newer first + } + if (lhs.snr != rhs.snr) { + return lhs.snr > rhs.snr; // stronger first when equally recent + } + return strcmp(lhs.pubkey_hex, rhs.pubkey_hex) < 0; +} + +// Build the neighbors-table JSON and hand it to the bridge, then reschedule. +void MyMesh::finishNeighborDiscover() { + getLocalScopes(self_scopes_buf, sizeof(self_scopes_buf)); + + char self_pubkey_hex[65]; + mesh::Utils::toHex(self_pubkey_hex, self_id.pub_key, PUB_KEY_SIZE); + + char origin[32]; + MQTTBridge::getEffectiveMqttOrigin(&_prefs, _cli.getObserverPrefs(), origin, sizeof(origin)); + + char timestamp[40]; + MQTTMessageBuilder::formatIsoTimestampForMqtt(getRTCClock()->getCurrentTime(), 0, nullptr, timestamp, sizeof(timestamp)); + + char pubkey_hex[MAX_NEIGHBOURS][65]; + MQTTMessageBuilder::NeighborsMessageEntry entries[MAX_NEIGHBOURS]; + uint32_t now_secs = getRTCClock()->getCurrentTime(); + + for (int i = 0; i < neighbor_discover_count; i++) { + auto& nb = neighbours[neighbor_discover[i].neighbour_idx]; + mesh::Utils::toHex(pubkey_hex[i], nb.id.pub_key, PUB_KEY_SIZE); + entries[i].pubkey_hex = pubkey_hex[i]; + entries[i].snr = nb.snr / 4.0f; + entries[i].heard_secs_ago = (nb.heard_timestamp > 0 && now_secs >= nb.heard_timestamp) + ? (now_secs - nb.heard_timestamp) : 0; + entries[i].scopes = neighbor_discover[i].scopes; + switch (neighbor_discover[i].status) { + case ND_RESPONDED: entries[i].status = "responded"; break; + case ND_SEND_FAILED: entries[i].status = "send_failed"; break; + default: entries[i].status = "timeout"; break; + } + } + + // insertion sort: most useful first (JSON builder drops the tail on overflow) + for (int i = 1; i < neighbor_discover_count; i++) { + MQTTMessageBuilder::NeighborsMessageEntry entry = entries[i]; + int j = i; + while (j > 0 && neighborPublishEntryComesBefore(entry, entries[j - 1])) { + entries[j] = entries[j - 1]; + j--; + } + entries[j] = entry; + } + +#if defined(ESP_PLATFORM) + char* json_buf = (char*)heap_caps_malloc(MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE, MALLOC_CAP_SPIRAM); +#else + char* json_buf = (char*)malloc(MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE); +#endif + if (!json_buf) { + neighbor_discover_active = false; + neighbor_discover_count = 0; + if (_cli.getObserverPrefs()->mqtt_neighbors_enabled) { + next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval); + } + return; + } + + JsonDocument doc; + int json_len = MQTTMessageBuilder::buildNeighborsMessage( + doc, origin, self_pubkey_hex, timestamp, self_scopes_buf, + entries, neighbor_discover_count, + json_buf, MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE); + + if (json_len > 0 && bridge) { + bridge->requestPublishNeighbors(json_buf, (size_t)json_len); + } + +#if defined(ESP_PLATFORM) + heap_caps_free(json_buf); +#else + free(json_buf); +#endif + + neighbor_discover_active = false; + neighbor_discover_count = 0; + if (_cli.getObserverPrefs()->mqtt_neighbors_enabled) { + next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval); + } +} + +// Advance the scope-query phase; publish once all entries resolve or the window +// times out (stragglers marked ND_TIMEOUT). +void MyMesh::loopNeighborDiscover() { + if (!neighbor_discover_active) return; + + bool all_done = true; + for (int i = 0; i < neighbor_discover_count; i++) { + if (neighbor_discover[i].status == ND_PENDING) { all_done = false; break; } + } + if (!all_done && !millisHasNowPassed(neighbor_discover_until)) return; + if (!all_done) { + for (int i = 0; i < neighbor_discover_count; i++) { + if (neighbor_discover[i].status == ND_PENDING) neighbor_discover[i].status = ND_TIMEOUT; + } + } + finishNeighborDiscover(); +} + +// Shared precondition for starting a discovery: PSRAM present + bridge running. +bool MyMesh::neighborDiscoverReady(char* reply) { +#if defined(ESP_PLATFORM) + if (!psramFound()) { strcpy(reply, "Err - PSRAM not available"); return false; } +#endif + if (!bridge || !bridge->isRunning()) { strcpy(reply, "Err - MQTT bridge not running"); return false; } + return true; +} + +// Snapshot the neighbor table into the overlay and fire one anon-regions query +// per heard neighbour; arm the 30s scope-query window. +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; + for (int i = 0; i < MAX_NEIGHBOURS; i++) { + if (neighbours[i].heard_timestamp > 0) { + neighbor_discover[neighbor_discover_count].neighbour_idx = (uint8_t)i; + neighbor_discover[neighbor_discover_count].scopes[0] = 0; + neighbor_discover[neighbor_discover_count].status = ND_PENDING; + uint32_t tag; + if (sendAnonRegionsReq(neighbours[i].id, tag)) { + neighbor_discover[neighbor_discover_count].tag = tag; + } else { + neighbor_discover[neighbor_discover_count].status = ND_SEND_FAILED; + } + neighbor_discover_count++; + } + } + + neighbor_discover_active = true; + neighbor_discover_until = futureMillis(NEIGHBOR_DISCOVER_TIMEOUT_MS); + + if (neighbor_discover_count == 0) { + finishNeighborDiscover(); + strcpy(reply, "OK - neighbor discover started (0 neighbors, self only)"); + } else { + sprintf(reply, "OK - neighbor discover started (%u neighbors)", (unsigned)neighbor_discover_count); + } + return true; +} +#endif // WITH_MQTT_NEIGHBORS + // To check if there is pending work bool MyMesh::hasPendingWork() const { #if defined(WITH_BRIDGE) diff --git a/examples/simple_repeater/MyMesh.h b/examples/simple_repeater/MyMesh.h index f1ec48d2..beb5c83e 100644 --- a/examples/simple_repeater/MyMesh.h +++ b/examples/simple_repeater/MyMesh.h @@ -147,6 +147,44 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks uint8_t _wc_slot_restart_mask = 0; #endif +#if defined(WITH_MQTT_NEIGHBORS) + // Neighbor-scope discovery: a snapshot of the neighbor table overlaid with an + // in-flight anon-regions query per neighbor, published to the MQTT neighbors + // topic once every neighbor has responded or the window times out. + enum NeighborDiscoverStatus : uint8_t { + ND_PENDING = 1, + ND_RESPONDED = 2, + ND_TIMEOUT = 3, + ND_SEND_FAILED = 4, + }; + struct NeighborDiscoverEntry { + uint8_t neighbour_idx; // index into neighbours[] + uint32_t tag; // anon-regions request tag we're waiting on + char scopes[96]; // scope names from the response + uint8_t status; // NeighborDiscoverStatus + }; + NeighborDiscoverEntry neighbor_discover[MAX_NEIGHBOURS]; + uint8_t neighbor_discover_count; + bool neighbor_discover_active; // scope-query phase in flight + bool neighbor_table_refresh_active; // zero-hop table refresh (stage 1) in flight + bool neighbor_table_refresh_periodic; // that refresh was kicked by the periodic timer + unsigned long neighbor_discover_until; // scope-query timeout deadline + unsigned long next_neighbors_publish; // periodic publish deadline (0 = fire ASAP) + 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(); + bool handleNeighborDiscoverResponse(int overlay_idx, const uint8_t* data, size_t len); + void getLocalScopes(char* buf, size_t len); + // Overlay peer indices are offset by this base so onPeerDataRecv can tell a + // discovery response apart from a normal ACL-client index. + static const int NEIGHBOR_DISCOVER_PEER_BASE = 1000; + static const unsigned long NEIGHBOR_DISCOVER_TIMEOUT_MS = 30000; +#endif + void putNeighbour(const mesh::Identity& id, uint32_t timestamp, float snr); uint8_t handleLoginReq(const mesh::Identity& sender, const uint8_t* secret, uint32_t sender_timestamp, const uint8_t* data, bool is_flood); uint8_t handleAnonRegionsReq(const mesh::Identity& sender, uint32_t sender_timestamp, const uint8_t* data); From 6911981573af62aac55ef04aedca975238cc7582 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 19 Jul 2026 22:38:10 -0700 Subject: [PATCH 5/7] feat(mqtt): add mqtt.neighbors CLI get/set (PSRAM-gated) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Port the neighbors CLI from mqtt-bridge-implementation-flex: - set mqtt.neighbors on|off - enable/disable periodic publishing - set mqtt.neighbors.interval - 12-336 hours (rejected, not clamped), stored ms - get mqtt.neighbors - "on"/"off" - get mqtt.neighbors.interval - "> hours ( ms)" (ceiling division) Both gated on WITH_MQTT_NEIGHBORS with a WITH_MQTT_BRIDGE stub replying "Err - not supported (requires PSRAM)" on non-PSRAM builds. Handlers only write prefs + savePrefs() — the mesh loop reads them live, so no bridge restart and no direct bridge/MyMesh call (matches flex). Token ordering preserved: SET tokens keep their trailing space so the shorter "mqtt.neighbors " can precede "mqtt.neighbors.interval "; GET tokens have no trailing space so the longer ".interval" is tested first. --- src/helpers/CommonCLI_Observer.cpp | 37 ++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/src/helpers/CommonCLI_Observer.cpp b/src/helpers/CommonCLI_Observer.cpp index 8e7c5426..5bed6837 100644 --- a/src/helpers/CommonCLI_Observer.cpp +++ b/src/helpers/CommonCLI_Observer.cpp @@ -263,6 +263,30 @@ bool CommonCLI::handleObserverSetCmd(uint32_t sender_timestamp, const char* conf } else { strcpy(reply, "Error: interval must be between 1-60 minutes"); } +#if defined(WITH_MQTT_NEIGHBORS) + } else if (memcmp(config, "mqtt.neighbors.interval ", 24) == 0) { + // Hours in, milliseconds stored. The 12-336h band keeps the interval under + // INT32_MAX so the mesh's wrap-safe signed-delta millis math stays valid. + uint32_t hours = _atoi(&config[24]); + if (hours >= MQTT_NEIGHBORS_MIN_INTERVAL_HOURS && hours <= MQTT_NEIGHBORS_MAX_INTERVAL_HOURS) { + _mqtt_prefs.mqtt_neighbors_interval = hours * 3600000UL; + savePrefs(); + sprintf(reply, "OK - neighbors interval set to %u hours (%lu ms)", (unsigned)hours, + (unsigned long)_mqtt_prefs.mqtt_neighbors_interval); + } else { + strcpy(reply, "Error: neighbors interval must be between 12-336 hours"); + } + } else if (memcmp(config, "mqtt.neighbors ", 15) == 0) { + // The mesh loop reads this live, so no bridge restart is needed; enabling it + // triggers a discovery on the next eligible loop pass. + _mqtt_prefs.mqtt_neighbors_enabled = memcmp(&config[15], "on", 2) == 0; + savePrefs(); + strcpy(reply, "OK"); +#elif defined(WITH_MQTT_BRIDGE) + } else if (memcmp(config, "mqtt.neighbors.interval ", 24) == 0 || + memcmp(config, "mqtt.neighbors ", 15) == 0) { + strcpy(reply, "Err - not supported (requires PSRAM)"); +#endif } else if (memcmp(config, "mqtt.ntp ", 9) == 0) { const char* host = &config[9]; while (*host == ' ') host++; @@ -738,6 +762,19 @@ bool CommonCLI::handleObserverGetCmd(uint32_t sender_timestamp, const char* conf } else if (memcmp(config, "mqtt.interval", 13) == 0) { uint32_t minutes = (_mqtt_prefs.mqtt_status_interval + 29999) / 60000; sprintf(reply, "> %u minutes (%lu ms)", minutes, (unsigned long)_mqtt_prefs.mqtt_status_interval); +#if defined(WITH_MQTT_NEIGHBORS) + // Longer token first: a bare "mqtt.neighbors" (14) would otherwise swallow + // "mqtt.neighbors.interval" since the GET tokens carry no trailing space. + } else if (memcmp(config, "mqtt.neighbors.interval", 23) == 0) { + uint32_t hours = (_mqtt_prefs.mqtt_neighbors_interval + 3599999) / 3600000; + sprintf(reply, "> %u hours (%lu ms)", (unsigned)hours, (unsigned long)_mqtt_prefs.mqtt_neighbors_interval); + } else if (memcmp(config, "mqtt.neighbors", 14) == 0) { + sprintf(reply, "> %s", _mqtt_prefs.mqtt_neighbors_enabled ? "on" : "off"); +#elif defined(WITH_MQTT_BRIDGE) + } else if (memcmp(config, "mqtt.neighbors.interval", 23) == 0 || + memcmp(config, "mqtt.neighbors", 14) == 0) { + strcpy(reply, "Err - not supported (requires PSRAM)"); +#endif } else if (memcmp(config, "mqtt.ntp.diag", 13) == 0 && (config[13] == '\0' || config[13] == ' ')) { #ifdef ESP_PLATFORM // Connectivity probe across all configured NTP servers; never updates the clock. From d6f8a871830e7eaaf48a5c4e13e4a57a31d8e1b1 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 19 Jul 2026 22:38:19 -0700 Subject: [PATCH 6/7] feat(webconfig): expose mqtt.neighbors controls in the web portal - Allow mqtt.neighbors and mqtt.neighbors.interval in the WebConfigKeys set-key allowlist (the CLI enforces the PSRAM guard; the stub reply handles non-PSRAM). - Emit neighbors + neighbors_interval (hours) in the WebConfigServer config JSON. - Add a "Publish neighbors" toggle and a "Neighbors interval (hours)" field (12-336) to the Publishing card, with getVal() cases in webui/index.html. - Cover both keys in test_webconfig_keys. WebConfigHtml.h is a gitignored build artifact regenerated by the pre-build hook from index.html, so it is not committed. Verified: test_webconfig_keys passes and the T_Beam_S3_Supreme observer_mqtt firmware builds [SUCCESS]. --- src/helpers/WebConfigKeys.h | 3 ++- src/helpers/esp32/WebConfigServer.cpp | 2 ++ test/test_webconfig_keys/test_webconfig_keys.cpp | 2 ++ webui/index.html | 10 +++++++++- 4 files changed, 15 insertions(+), 2 deletions(-) diff --git a/src/helpers/WebConfigKeys.h b/src/helpers/WebConfigKeys.h index 0b1f088c..7579cc12 100644 --- a/src/helpers/WebConfigKeys.h +++ b/src/helpers/WebConfigKeys.h @@ -23,7 +23,8 @@ static const char* const WC_ALLOWED_SET_KEYS[] = { // MQTTPrefs (WiFi / MQTT / misc observer) "wifi.ssid", "wifi.pwd", "wifi.powersave", "mqtt.origin", "mqtt.iata", "mqtt.status", "mqtt.packets", "mqtt.raw", - "mqtt.tx", "mqtt.rx", "mqtt.interval", "mqtt.ntp", "mqtt.owner", "mqtt.email", + "mqtt.tx", "mqtt.rx", "mqtt.interval", "mqtt.neighbors", "mqtt.neighbors.interval", + "mqtt.ntp", "mqtt.owner", "mqtt.email", "timezone", "timezone.offset", "snmp", "snmp.community", }; static const char* const WC_ALLOWED_SLOT_KEYS[] = { diff --git a/src/helpers/esp32/WebConfigServer.cpp b/src/helpers/esp32/WebConfigServer.cpp index 5603a55d..8b4edb6a 100644 --- a/src/helpers/esp32/WebConfigServer.cpp +++ b/src/helpers/esp32/WebConfigServer.cpp @@ -591,6 +591,8 @@ void WebConfigServer::handleConfigGet(AsyncWebServerRequest* req) { : _obs->mqtt_tx_enabled == 1 ? "on" : "off"; mqtt["rx"] = (bool)_obs->mqtt_rx_enabled; mqtt["interval"] = _obs->mqtt_status_interval / 60000; // CLI takes minutes + mqtt["neighbors"] = (bool)_obs->mqtt_neighbors_enabled; + mqtt["neighbors_interval"] = _obs->mqtt_neighbors_interval / 3600000UL; // CLI takes hours mqtt["timezone"] = (const char*)_obs->timezone_string; mqtt["timezone_offset"] = _obs->timezone_offset; mqtt["ntp"] = (const char*)_obs->mqtt_ntp_server; diff --git a/test/test_webconfig_keys/test_webconfig_keys.cpp b/test/test_webconfig_keys/test_webconfig_keys.cpp index 098463fc..b2ef4fe4 100644 --- a/test/test_webconfig_keys/test_webconfig_keys.cpp +++ b/test/test_webconfig_keys/test_webconfig_keys.cpp @@ -12,6 +12,8 @@ TEST(WebConfigKeys, AllowsKnownScalarKeys) { EXPECT_TRUE(wcIsAllowedSetKey("repeat")); EXPECT_TRUE(wcIsAllowedSetKey("wifi.ssid")); EXPECT_TRUE(wcIsAllowedSetKey("mqtt.iata")); + EXPECT_TRUE(wcIsAllowedSetKey("mqtt.neighbors")); + EXPECT_TRUE(wcIsAllowedSetKey("mqtt.neighbors.interval")); EXPECT_TRUE(wcIsAllowedSetKey("snmp.community")); EXPECT_TRUE(wcIsAllowedSetKey("timezone.offset")); } diff --git a/webui/index.html b/webui/index.html index cb6e5e82..b7b65474 100644 --- a/webui/index.html +++ b/webui/index.html @@ -313,12 +313,18 @@ canvas{width:100%;height:56px;display:block}
Received packetsReport packets heard over the air
+
Publish neighborsPeriodic neighbor table + scopes (PSRAM boards only) +
Report packets this node sends.
+
+
+
How often to publish the neighbor table (12-336, default 24).
+

Servers

@@ -520,7 +526,9 @@ function cfgVal(k){ // map a `set` key to its current value string, from st.cfg case"mqtt.origin":return q.origin;case"mqtt.iata":return q.iata; case"mqtt.status":return q.status?"on":"off";case"mqtt.packets":return q.packets?"on":"off"; case"mqtt.raw":return q.raw?"on":"off";case"mqtt.tx":return q.tx;case"mqtt.rx":return q.rx?"on":"off"; - case"mqtt.interval":return String(q.interval);case"mqtt.ntp":return q.ntp; + case"mqtt.interval":return String(q.interval); + case"mqtt.neighbors":return q.neighbors?"on":"off";case"mqtt.neighbors.interval":return String(q.neighbors_interval); + case"mqtt.ntp":return q.ntp; case"mqtt.owner":return q.owner;case"mqtt.email":return q.email; case"timezone":return q.timezone;case"timezone.offset":return String(q.timezone_offset); case"snmp":return q.snmp?"on":"off";case"snmp.community":return q.snmp_community; From 0ae6aa4df257d4105a3d9133100a4eb1d57ef982 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 19 Jul 2026 22:38:27 -0700 Subject: [PATCH 7/7] docs(mqtt): document the neighbors feature - MQTT_IMPLEMENTATION.md: neighbors topic, JSON message example, get/set command lists, and the MeshRank packets-only note. - docs/cli_commands.md: mqtt.neighbors / mqtt.neighbors.interval get/set and the discover.scopes command, with PSRAM-only notes. - MQTT_INTERNALS.md: the two-stage discovery + Core1->Core0 publish handoff, peer-overlay decryption, buffer sizing, and MeshRank exclusion. --- MQTT_IMPLEMENTATION.md | 32 ++++++++++++++++++++++++++++ MQTT_INTERNALS.md | 31 ++++++++++++++++++++++++++++ docs/cli_commands.md | 47 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 110 insertions(+) diff --git a/MQTT_IMPLEMENTATION.md b/MQTT_IMPLEMENTATION.md index 0da06dc5..ba0fe5d2 100644 --- a/MQTT_IMPLEMENTATION.md +++ b/MQTT_IMPLEMENTATION.md @@ -382,6 +382,8 @@ These settings apply across all MQTT slots: - `get mqtt.rx` - Get RX packet uplinking setting (on/off) - `get mqtt.tx` - Get TX packet uplinking setting (on/off/advert) - `get mqtt.interval` - Get status publish interval +- `get mqtt.neighbors` - Get periodic neighbors publishing setting (on/off; PSRAM only) +- `get mqtt.neighbors.interval` - Get neighbors publish interval in hours (PSRAM only) - `get mqtt.ntp` - Get effective NTP server hostname - `get mqtt.ntp.diag` - Probe every configured NTP server for connectivity (does not change the clock; serial console shows each server's reported time, LoRa shows a compact ` ok|fail` list) - `get mqtt.owner` - Get owner public key (serial console only) @@ -399,6 +401,8 @@ These settings apply across all MQTT slots: - `advert` - Uplink only this node's own advert packets (self-originated) - `off` - Disable TX packet uplinking - `set mqtt.interval ` - Set status publish interval (1-60 minutes) +- `set mqtt.neighbors on|off` - Enable/disable periodic neighbors publishing (PSRAM only; read live, no restart) +- `set mqtt.neighbors.interval ` - Set neighbors publish interval (12-336 hours, default 24; PSRAM only) - `set mqtt.ntp ` - Set custom NTP server (validated with immediate sync); `none` reverts to default - `set mqtt.owner <64-hex-char-public-key>` - Set owner public key - `set mqtt.email ` - Set owner email address @@ -584,6 +588,12 @@ Full packet data with RF characteristics and metadata. ### Raw Topic: `meshcore/{IATA}/{DEVICE_PUBLIC_KEY}/raw` Minimal raw packet data for map integration. +### Neighbors Topic: `meshcore/{IATA}/{DEVICE_PUBLIC_KEY}/neighbors` +Periodic snapshot of this node's zero-hop neighbor table plus each neighbor's +region scopes (PSRAM boards only; disabled by default). Published non-retained at +QoS 1 on the interval set by `mqtt.neighbors.interval` (12–336 h, default 24 h). +Like status/raw, this topic is **not** sent to MeshRank slots (packets-only contract). + **Note**: `{DEVICE_PUBLIC_KEY}` is the device's public key in hexadecimal format (64 characters). ## JSON Message Formats @@ -662,6 +672,28 @@ Minimal raw packet data for map integration. } ``` +### Neighbors Message +```json +{ + "timestamp": "2024-01-01T12:00:00.000000+00:00", + "origin": "MeshCore-HOWL", + "origin_id": "A1B2C3D4E5F67890...", + "self": { "scopes": "DEN,APRS" }, + "neighbors": [ + { + "pubkey": "0011223344556677...", + "snr": 9.75, + "heard_secs_ago": 42, + "scopes": "DEN,APRS", + "status": "responded" + } + ] +} +``` +Entries are ordered most- to least-useful (most recently heard, then stronger +SNR); the tail is dropped if the payload would exceed the 10 KB publish buffer. +`status` is `responded`, `timeout`, or `send_failed` per neighbor. + ## Key Features ### Slot-Based Preset System diff --git a/MQTT_INTERNALS.md b/MQTT_INTERNALS.md index 426d4b9c..83e10a51 100644 --- a/MQTT_INTERNALS.md +++ b/MQTT_INTERNALS.md @@ -57,6 +57,37 @@ scheduled time are expired at dequeue, so under throttle the queue holds only fr traffic and admin responses reach the trickle of TX budget. Non-observer builds keep the upstream pool behavior. +### Neighbors publication path (PSRAM only) + +Periodic neighbors publishing is gated on `WITH_MQTT_NEIGHBORS` +(`defined(BOARD_HAS_PSRAM) && defined(MAX_NEIGHBOURS) && MAX_NEIGHBOURS > 0`, +defined in `MQTTBridge.h`). It spans two subsystems and two cores: + +- **Mesh side (Core 1), `MyMesh`**: the `loop()` runs a two-stage refresh driven by + `mqtt_neighbors_interval`. Stage 1 sends a zero-hop `sendNodeDiscoverReq()` and waits + out its 60 s collection window to refresh `neighbours[]`. Stage 2 + (`startNeighborDiscover`) fires one anon-regions scope query per heard neighbor, + overlaying them onto the peer-index space at `NEIGHBOR_DISCOVER_PEER_BASE` so their + `PAYLOAD_TYPE_RESPONSE` packets decrypt via `searchPeersByHash`/`getPeerSharedSecret`/ + `onPeerDataRecv` even when the neighbor is not an ACL client. After all responses land + or a 30 s window expires, `finishNeighborDiscover()` builds the JSON with + `MQTTMessageBuilder::buildNeighborsMessage` into a transient PSRAM buffer and hands it + to the bridge. +- **Bridge side, handoff**: `requestPublishNeighbors(json, len)` (Core 1) memcpys into a + persistent ~10 KB PSRAM buffer (`NEIGHBORS_JSON_BUFFER_SIZE`) and sets + `_neighbors_publish_pending` with a release store; the MQTT task (`mqttTaskLoop`, Core 0) + consumes it with an acquire load, calls `publishNeighbors()`, and clears the flag. A + second snapshot is dropped while one is in flight. `publishNeighbors()` sends QoS 1, + retain = `preset->allow_retain` (custom slots non-retained). MeshRank slots are skipped + (the topic router rejects non-packets for MeshRank). +- **Status reporting**: `MyMesh` reports the schedule each loop via + `setNeighborsSchedule(phase, secs)`; `formatMqttStatusReply` renders it as the trailing + `nbr: /` field in `get mqtt.status` while the feature is enabled. + +The JSON builder lives in the pure, host-tested `MQTTPayloadBuilder` +(`test/test_mqtt_payload_builder`); the topic type in `MQTTTopicRouter` +(`test/test_mqtt_topic_router`). The mesh↔bridge orchestration above is on-target only. + ### `/mqtt_prefs` file format `/mqtt_prefs` is written with an 8-byte `MQTTPrefsHeader` (`magic`, `version`, diff --git a/docs/cli_commands.md b/docs/cli_commands.md index b0badbcd..c4bb5d9d 100644 --- a/docs/cli_commands.md +++ b/docs/cli_commands.md @@ -113,6 +113,20 @@ This document provides an overview of CLI commands that can be sent to MeshCore --- +### Discover neighbor scopes (MQTT observer, PSRAM only) + +Refreshes the zero-hop neighbor table, then queries each neighbor for its region +scopes and publishes the assembled table to the MQTT `neighbors` topic once. + +**Usage:** +- `discover.scopes` + +**Note:** Requires a PSRAM board with the MQTT bridge running. On non-PSRAM MQTT +builds it replies `Err - not supported (requires PSRAM)`. If a `discover.neighbors` +refresh is already in flight, the scope pass is queued behind it. + +--- + ## Statistics ### Clear Stats @@ -1076,6 +1090,39 @@ region save --- +#### View or change periodic neighbors publishing (MQTT observer, PSRAM only) +**Usage:** +- `get mqtt.neighbors` +- `set mqtt.neighbors ` + +**Parameters:** +- `on`: periodically discover neighbor scopes and publish the neighbor table to the `neighbors` topic +- `off`: disable periodic neighbors publishing + +**Default:** `off` + +> **Note:** Requires a PSRAM board. On non-PSRAM MQTT builds this replies +> `Err - not supported (requires PSRAM)`. The setting is read live by the mesh +> loop — no restart required; enabling it triggers a discovery on the next pass. +> While enabled, `get mqtt.status` gains a trailing `nbr: /` field +> (time to next publish, and how the last publish went). + +--- + +#### View or change the neighbors publish interval (MQTT observer, PSRAM only) +**Usage:** +- `get mqtt.neighbors.interval` +- `set mqtt.neighbors.interval ` + +**Parameters:** +- `hours`: how often to publish the neighbor table (12–336, default 24) + +**Default:** `24` (hours) + +> **Note:** Out-of-range values are rejected (not clamped). Requires a PSRAM board. + +--- + #### View or change the NTP server (MQTT observer only) **Usage:** - `get mqtt.ntp`