diff --git a/platformio.ini b/platformio.ini index 102d486c..29147345 100644 --- a/platformio.ini +++ b/platformio.ini @@ -172,5 +172,7 @@ test_build_src = yes build_src_filter = -<*> +<../src/Utils.cpp> + +<../src/helpers/MQTTPayloadBuilder.cpp> lib_deps = google/googletest @ 1.17.0 + bblanchon/ArduinoJson @ 7.4.3 diff --git a/src/helpers/MQTTConnectionPolicy.h b/src/helpers/MQTTConnectionPolicy.h new file mode 100644 index 00000000..42e263a9 --- /dev/null +++ b/src/helpers/MQTTConnectionPolicy.h @@ -0,0 +1,135 @@ +#pragma once + +#include + +// Pure timing and state-transition policy used by MQTTBridge's connection +// maintenance loop. Keeping these decisions independent of Arduino, WiFi, and +// the MQTT client lets host tests exercise the exact production policy with a +// deterministic clock. +namespace MQTTConnectionPolicy { + +static const uint32_t kReconnectGuardMs = 15000UL; +static const uint32_t kStableResetMs = 120000UL; +static const uint32_t kCircuitBreakerProbeMs = 1800000UL; +static const uint32_t kRenewalThrottleMs = 60000UL; +static const uint32_t kSlotStaggerMs = 3000UL; +static const uint8_t kMaxFailuresAtMaxBackoff = 3; +static const uint32_t kDefaultJwtLifetimeSecs = 86400UL; +static const uint32_t kMaxJwtStaggerSecs = 300UL; +static const uint32_t kMinimumValidEpoch = 1000000000UL; +static const uint32_t kJwtClockThreshold = 1735689600UL; // 2025-01-01 UTC + +// Unsigned subtraction is intentionally used: it is the standard millis() +// idiom and remains correct across a single 32-bit counter rollover. +static inline uint32_t elapsedMs(uint32_t now, uint32_t then) { + return now - then; +} + +static inline bool reconnectGuardActive(uint32_t now, uint32_t last_reconnect) { + return elapsedMs(now, last_reconnect) < kReconnectGuardMs; +} + +static inline bool stableConnection(uint32_t now, uint32_t connected_at) { + return connected_at != 0 && elapsedMs(now, connected_at) >= kStableResetMs; +} + +static inline uint32_t reconnectBackoffMs(uint8_t reconnect_backoff) { + static const uint32_t kBackoffMs[] = { + 10000UL, 30000UL, 60000UL, 120000UL, 300000UL + }; + const uint8_t index = reconnect_backoff < 5 ? reconnect_backoff : 4; + return kBackoffMs[index]; +} + +static inline uint32_t reconnectDelayMs(uint8_t reconnect_backoff, uint8_t slot_index) { + return reconnectBackoffMs(reconnect_backoff) + + static_cast(slot_index) * kSlotStaggerMs; +} + +static inline bool reconnectDue(uint32_t now, uint32_t last_attempt, + uint8_t reconnect_backoff, uint8_t slot_index) { + return elapsedMs(now, last_attempt) >= reconnectDelayMs(reconnect_backoff, slot_index); +} + +struct BackoffAdvance { + uint8_t reconnect_backoff; + uint8_t max_backoff_failures; + bool circuit_breaker_tripped; + bool should_reconnect; +}; + +// Advance the ladder immediately before a due reconnect. The first visit to +// the 300-second rung changes level 4 to the saturated marker 5. Three later +// failures at that rung trip the breaker; the third does not launch another +// connection attempt. +static inline BackoffAdvance advanceBackoff(uint8_t reconnect_backoff, + uint8_t max_backoff_failures) { + BackoffAdvance result = { + reconnect_backoff, max_backoff_failures, false, true + }; + if (result.reconnect_backoff < 5) { + result.reconnect_backoff++; + return result; + } + + if (result.max_backoff_failures < UINT8_MAX) { + result.max_backoff_failures++; + } + if (result.max_backoff_failures >= kMaxFailuresAtMaxBackoff) { + result.circuit_breaker_tripped = true; + result.should_reconnect = false; + } + return result; +} + +static inline bool circuitBreakerProbeDue(uint32_t now, uint32_t last_attempt) { + return elapsedMs(now, last_attempt) >= kCircuitBreakerProbeMs; +} + +// Each later slot expires up to five percent of the base lifetime earlier, +// capped at five minutes per slot. Runtime slot indexes are bounded by the +// persisted MQTT slot count; the final clamp also prevents underflow if this +// helper is used with unexpected input. +static inline uint32_t jwtLifetimeSecs(uint32_t base_lifetime, uint8_t slot_index) { + uint32_t per_slot_stagger = base_lifetime / 20UL; + if (per_slot_stagger > kMaxJwtStaggerSecs) { + per_slot_stagger = kMaxJwtStaggerSecs; + } + uint64_t stagger = static_cast(slot_index) * per_slot_stagger; + if (stagger > base_lifetime) { + stagger = base_lifetime; + } + return base_lifetime - static_cast(stagger); +} + +static inline uint32_t renewalBufferSecs(uint32_t lifetime_secs) { + uint32_t buffer = lifetime_secs / 10UL; + if (buffer < 60UL) buffer = 60UL; + if (buffer > 300UL) buffer = 300UL; + return buffer; +} + +static inline bool tokenNeedsRenewal(bool time_synced, uint32_t current_time, + uint32_t token_expires_at, + uint32_t renewal_buffer_secs) { + if (!time_synced) { + return token_expires_at == 0; + } + if (token_expires_at < kMinimumValidEpoch) { + return true; + } + if (current_time >= token_expires_at) { + return true; + } + return current_time >= token_expires_at - renewal_buffer_secs; +} + +static inline bool renewalAttemptAllowed(uint32_t now, uint32_t last_attempt) { + return elapsedMs(now, last_attempt) >= kRenewalThrottleMs; +} + +static inline bool jwtClockAvailable(bool ntp_synced, uint32_t current_time) { + return ntp_synced || current_time >= kJwtClockThreshold; +} + +} // namespace MQTTConnectionPolicy diff --git a/src/helpers/MQTTMessageBuilder.cpp b/src/helpers/MQTTMessageBuilder.cpp index a29b3d96..fb0d4399 100644 --- a/src/helpers/MQTTMessageBuilder.cpp +++ b/src/helpers/MQTTMessageBuilder.cpp @@ -1,4 +1,5 @@ #include "MQTTMessageBuilder.h" +#include "MQTTPayloadBuilder.h" #include #include #include @@ -53,66 +54,11 @@ int MQTTMessageBuilder::buildStatusMessage( int packets_received, const char* repeat ) { - // doc is provided by the caller (heap-allocated DynamicJsonDocument in MQTTBridge), - // keeping this 768-byte scratch space off the MQTT task stack. - doc.clear(); - JsonObject root = doc.to(); - - root["status"] = status; - root["timestamp"] = timestamp; - root["origin"] = origin; - root["origin_id"] = origin_id; - root["model"] = model; - root["firmware_version"] = firmware_version; - root["radio"] = radio; - root["client_version"] = client_version; - if (repeat != nullptr) { - root["repeat"] = repeat; - } - - // Add stats object if any stats are provided - if (battery_mv >= 0 || uptime_secs >= 0 || errors >= 0 || queue_len >= 0 || - noise_floor > -999 || tx_air_secs >= 0 || rx_air_secs >= 0 || recv_errors >= 0 || - internal_heap >= 0 || packets_sent >= 0 || packets_received >= 0) { - JsonObject stats = root.createNestedObject("stats"); - - if (battery_mv >= 0) { - stats["battery_mv"] = battery_mv; - } - if (uptime_secs >= 0) { - stats["uptime_secs"] = uptime_secs; - } - if (packets_sent >= 0) { - stats["packets_sent"] = packets_sent; - } - if (packets_received >= 0) { - stats["packets_received"] = packets_received; - } - if (errors >= 0) { - stats["errors"] = errors; - } - if (queue_len >= 0) { - stats["queue_len"] = queue_len; - } - if (noise_floor > -999) { - stats["noise_floor"] = noise_floor; - } - if (tx_air_secs >= 0) { - stats["tx_air_secs"] = tx_air_secs; - } - if (rx_air_secs >= 0) { - stats["rx_air_secs"] = rx_air_secs; - } - if (recv_errors >= 0) { - stats["recv_errors"] = recv_errors; - } - if (internal_heap >= 0) { - stats["internal_heap"] = internal_heap; - } - } - - size_t len = serializeJson(root, buffer, buffer_size); - return (len > 0 && len < buffer_size) ? len : 0; + return MQTTPayloadBuilder::buildStatusMessage( + doc, origin, origin_id, model, firmware_version, radio, client_version, + status, timestamp, buffer, buffer_size, battery_mv, uptime_secs, errors, + queue_len, noise_floor, tx_air_secs, rx_air_secs, recv_errors, internal_heap, + packets_sent, packets_received, repeat); } int MQTTMessageBuilder::buildPacketMessage( @@ -138,71 +84,10 @@ int MQTTMessageBuilder::buildPacketMessage( char* buffer, size_t buffer_size ) { - // doc is provided by the caller (heap-allocated DynamicJsonDocument in MQTTBridge), - // keeping this 2048-byte scratch space off the MQTT task stack. - doc.clear(); - JsonObject root = doc.to(); - - // Format numeric values as strings to avoid String object allocations - char len_str[16]; - char packet_type_str[16]; - char payload_len_str[16]; - char snr_str[16]; - char rssi_str[16]; - char score_str[16]; - - snprintf(len_str, sizeof(len_str), "%d", len); - snprintf(packet_type_str, sizeof(packet_type_str), "%d", packet_type); - snprintf(payload_len_str, sizeof(payload_len_str), "%d", payload_len); - snprintf(snr_str, sizeof(snr_str), "%.1f", snr); - snprintf(rssi_str, sizeof(rssi_str), "%d", rssi); - - root["timestamp"] = timestamp; - root["hash"] = hash; - root["origin"] = origin; - root["type"] = "PACKET"; - root["direction"] = direction; - root["time"] = time; - root["date"] = date; - root["len"] = len_str; - root["packet_type"] = packet_type_str; - root["route"] = route; - root["payload_len"] = payload_len_str; - root["raw"] = raw; - root["origin_id"] = origin_id; - // SNR and RSSI are only meaningful for RX packets (received from radio) - if (strcmp(direction, "rx") == 0) { - root["SNR"] = snr_str; - root["RSSI"] = rssi_str; - // Firmware's rebroadcast "score" for this RX packet, scaled x1000 to match the - // integer form printed in the serial RX log (see Dispatcher::checkRecv()). - if (!isnan(score)) { - snprintf(score_str, sizeof(score_str), "%d", (int)(score * 1000)); - root["score"] = score_str; - } - } - - // Routing path as an array of lowercase hex hop tokens, one element per hop - // (e.g. ["aa","bb","cc"], or ["aaaa","bbbb"] for multi-byte hashes). This matches - // meshcore-packet-capture's _split_path_hops() representation. - if (path_bytes && path_hop_count > 0 && path_hash_size > 0) { - JsonArray path_arr = root.createNestedArray("path"); - char hop_hex[2 * 4 + 1]; // hop hash is 1-4 bytes -> up to 8 hex chars + null - for (int i = 0; i < path_hop_count; i++) { - size_t pos = 0; - for (int b = 0; b < path_hash_size && b < 4; b++) { - size_t idx = (size_t)i * path_hash_size + b; - if (idx >= MAX_PATH_SIZE) break; - snprintf(hop_hex + pos, 3, "%02x", path_bytes[idx]); - pos += 2; - } - hop_hex[pos] = '\0'; - path_arr.add(hop_hex); // char[] (non-const) -> ArduinoJson copies the string - } - } - - size_t json_len = serializeJson(root, buffer, buffer_size); - return (json_len > 0 && json_len < buffer_size) ? json_len : 0; + return MQTTPayloadBuilder::buildPacketMessage( + doc, origin, origin_id, timestamp, direction, time, date, len, packet_type, + route, payload_len, raw, snr, rssi, score, hash, path_bytes, path_hop_count, + path_hash_size, MAX_PATH_SIZE, buffer, buffer_size); } int MQTTMessageBuilder::buildRawMessage( @@ -213,18 +98,8 @@ int MQTTMessageBuilder::buildRawMessage( char* buffer, size_t buffer_size ) { - // Use StaticJsonDocument to avoid heap fragmentation (fixed-size stack allocation) - StaticJsonDocument<512> doc; - JsonObject root = doc.to(); - - root["origin"] = origin; - root["origin_id"] = origin_id; - root["timestamp"] = timestamp; - root["type"] = "RAW"; - root["data"] = raw; - - size_t len = serializeJson(root, buffer, buffer_size); - return (len > 0 && len < buffer_size) ? len : 0; + return MQTTPayloadBuilder::buildRawMessage( + origin, origin_id, timestamp, raw, buffer, buffer_size); } int MQTTMessageBuilder::buildPacketJSON( @@ -435,4 +310,4 @@ void MQTTMessageBuilder::packetToHex(mesh::Packet* packet, char* hex, size_t hex // Convert serialized packet to hex bytesToHex(raw_buf, raw_len, hex, hex_size); -} \ No newline at end of file +} diff --git a/src/helpers/MQTTPacketQueuePolicy.h b/src/helpers/MQTTPacketQueuePolicy.h new file mode 100644 index 00000000..74eabf47 --- /dev/null +++ b/src/helpers/MQTTPacketQueuePolicy.h @@ -0,0 +1,105 @@ +#pragma once + +#include +#include + +// Pure queue/backpressure policy shared by the FreeRTOS and circular-buffer +// MQTT packet queues. Keeping the timing and retry decisions here makes the +// production behavior deterministic under host tests without mocking either +// queue implementation or the MQTT client. +namespace MQTTPacketQueuePolicy { + +static const uint32_t kDisconnectedStaleMs = 300000UL; +static const size_t kBacklogThreshold = 5; +static const uint8_t kGentleDrainCount = 1; +static const uint8_t kBurstDrainCount = 5; +static const uint32_t kGentleDrainBudgetMs = 30UL; +static const uint32_t kBurstDrainBudgetMs = 100UL; +static const uint8_t kMaxQos0RetryAttempts = 3; +static const uint32_t kRetryDelayBaseMs = 300UL; +static const uint32_t kRetryDelayJitterMs = 200UL; + +// Unsigned subtraction is the standard millis() idiom and remains correct +// across one 32-bit counter rollover. +static inline uint32_t elapsedMs(uint32_t now, uint32_t then) { + return now - then; +} + +enum class EnqueueAction : uint8_t { + Enqueue, + EvictOldestThenEnqueue, + Reject +}; + +static inline EnqueueAction enqueueAction(size_t queue_count, size_t capacity) { + if (capacity == 0) return EnqueueAction::Reject; + return queue_count >= capacity + ? EnqueueAction::EvictOldestThenEnqueue + : EnqueueAction::Enqueue; +} + +// disconnected_since == 0 means tracking has not started. The bridge records +// the first disconnected observation and asks this helper on later cycles. +static inline bool shouldFlushDisconnected(uint32_t now, + uint32_t disconnected_since, + uint32_t stale_ms = kDisconnectedStaleMs) { + return disconnected_since != 0 && elapsedMs(now, disconnected_since) >= stale_ms; +} + +struct DrainBudget { + uint8_t max_packets; + uint32_t max_time_ms; +}; + +static inline DrainBudget drainBudget(size_t queue_count) { + if (queue_count > kBacklogThreshold) { + return {kBurstDrainCount, kBurstDrainBudgetMs}; + } + return {kGentleDrainCount, kGentleDrainBudgetMs}; +} + +static inline bool drainTimeAvailable(uint32_t now, uint32_t started_at, + uint32_t budget_ms) { + // Preserve the bridge's inclusive boundary: work may begin at exactly the + // configured limit, but not one millisecond later. + return elapsedMs(now, started_at) <= budget_ms; +} + +// retry_attempts distinguishes an unscheduled packet from a scheduled retry +// whose deadline wrapped to exactly zero. Deadlines are always less than +// 500 ms away, so the half-range comparison is unambiguous. +static inline bool retryReady(uint32_t now, uint32_t next_retry_ms, + uint8_t retry_attempts) { + if (retry_attempts == 0) return true; + return elapsedMs(now, next_retry_ms) < 0x80000000UL; +} + +enum class RetryAction : uint8_t { + Complete, + Schedule, + Drop +}; + +struct RetryDecision { + RetryAction action; + uint8_t retry_attempts; + uint32_t delay_ms; + uint32_t next_retry_ms; +}; + +static inline RetryDecision retryDecision(bool any_published, + uint8_t retry_attempts, + uint32_t now) { + if (any_published) { + return {RetryAction::Complete, retry_attempts, 0, 0}; + } + if (retry_attempts >= kMaxQos0RetryAttempts) { + return {RetryAction::Drop, retry_attempts, 0, 0}; + } + + const uint32_t delay = kRetryDelayBaseMs + (now % kRetryDelayJitterMs); + return {RetryAction::Schedule, static_cast(retry_attempts + 1), + delay, now + delay}; +} + +} // namespace MQTTPacketQueuePolicy diff --git a/src/helpers/MQTTPayloadBuilder.cpp b/src/helpers/MQTTPayloadBuilder.cpp new file mode 100644 index 00000000..260fac5f --- /dev/null +++ b/src/helpers/MQTTPayloadBuilder.cpp @@ -0,0 +1,186 @@ +#include "MQTTPayloadBuilder.h" + +#include +#include +#include + +namespace { + +static int serializeComplete(JsonObject root, char* buffer, size_t buffer_size) { + if (!buffer || buffer_size == 0) return 0; + + size_t written = serializeJson(root, buffer, buffer_size); + // Preserve MQTTMessageBuilder's existing success criterion while clearing + // ArduinoJson's truncated prefix on failure. Callers publish only a positive + // return value, and now a failed buffer cannot be mistaken for complete JSON. + if (written == 0 || written >= buffer_size) { + buffer[0] = '\0'; + return 0; + } + return static_cast(written); +} + +} // namespace + +int MQTTPayloadBuilder::buildStatusMessage( + JsonDocument& doc, + const char* origin, + const char* origin_id, + const char* model, + const char* firmware_version, + const char* radio, + const char* client_version, + const char* status, + const char* timestamp, + char* buffer, + size_t buffer_size, + int battery_mv, + int uptime_secs, + int errors, + int queue_len, + int noise_floor, + int tx_air_secs, + int rx_air_secs, + int recv_errors, + int internal_heap, + int packets_sent, + int packets_received, + const char* repeat +) { + doc.clear(); + JsonObject root = doc.to(); + + root["status"] = status; + root["timestamp"] = timestamp; + root["origin"] = origin; + root["origin_id"] = origin_id; + root["model"] = model; + root["firmware_version"] = firmware_version; + root["radio"] = radio; + root["client_version"] = client_version; + if (repeat != nullptr) { + root["repeat"] = repeat; + } + + if (battery_mv >= 0 || uptime_secs >= 0 || errors >= 0 || queue_len >= 0 || + noise_floor > -999 || tx_air_secs >= 0 || rx_air_secs >= 0 || recv_errors >= 0 || + internal_heap >= 0 || packets_sent >= 0 || packets_received >= 0) { + JsonObject stats = root["stats"].to(); + + if (battery_mv >= 0) stats["battery_mv"] = battery_mv; + if (uptime_secs >= 0) stats["uptime_secs"] = uptime_secs; + if (packets_sent >= 0) stats["packets_sent"] = packets_sent; + if (packets_received >= 0) stats["packets_received"] = packets_received; + if (errors >= 0) stats["errors"] = errors; + if (queue_len >= 0) stats["queue_len"] = queue_len; + if (noise_floor > -999) stats["noise_floor"] = noise_floor; + if (tx_air_secs >= 0) stats["tx_air_secs"] = tx_air_secs; + if (rx_air_secs >= 0) stats["rx_air_secs"] = rx_air_secs; + if (recv_errors >= 0) stats["recv_errors"] = recv_errors; + if (internal_heap >= 0) stats["internal_heap"] = internal_heap; + } + + return serializeComplete(root, buffer, buffer_size); +} + +int MQTTPayloadBuilder::buildPacketMessage( + JsonDocument& doc, + const char* origin, + const char* origin_id, + const char* timestamp, + const char* direction, + const char* time, + const char* date, + int len, + int packet_type, + const char* route, + int payload_len, + const char* raw, + float snr, + int rssi, + float score, + const char* hash, + const uint8_t* path_bytes, + int path_hop_count, + int path_hash_size, + size_t max_path_bytes, + char* buffer, + size_t buffer_size +) { + doc.clear(); + JsonObject root = doc.to(); + + char len_str[16]; + char packet_type_str[16]; + char payload_len_str[16]; + char snr_str[16]; + char rssi_str[16]; + char score_str[16]; + + snprintf(len_str, sizeof(len_str), "%d", len); + snprintf(packet_type_str, sizeof(packet_type_str), "%d", packet_type); + snprintf(payload_len_str, sizeof(payload_len_str), "%d", payload_len); + snprintf(snr_str, sizeof(snr_str), "%.1f", snr); + snprintf(rssi_str, sizeof(rssi_str), "%d", rssi); + + root["timestamp"] = timestamp; + root["hash"] = hash; + root["origin"] = origin; + root["type"] = "PACKET"; + root["direction"] = direction; + root["time"] = time; + root["date"] = date; + root["len"] = len_str; + root["packet_type"] = packet_type_str; + root["route"] = route; + root["payload_len"] = payload_len_str; + root["raw"] = raw; + root["origin_id"] = origin_id; + + if (direction && strcmp(direction, "rx") == 0) { + root["SNR"] = snr_str; + root["RSSI"] = rssi_str; + if (!isnan(score)) { + snprintf(score_str, sizeof(score_str), "%d", static_cast(score * 1000)); + root["score"] = score_str; + } + } + + if (path_bytes && path_hop_count > 0 && path_hash_size > 0) { + JsonArray path_arr = root["path"].to(); + char hop_hex[2 * 4 + 1]; + for (int i = 0; i < path_hop_count; i++) { + size_t pos = 0; + for (int b = 0; b < path_hash_size && b < 4; b++) { + size_t idx = static_cast(i) * path_hash_size + b; + if (idx >= max_path_bytes) break; + snprintf(hop_hex + pos, 3, "%02x", path_bytes[idx]); + pos += 2; + } + hop_hex[pos] = '\0'; + path_arr.add(hop_hex); + } + } + + return serializeComplete(root, buffer, buffer_size); +} + +int MQTTPayloadBuilder::buildRawMessage( + const char* origin, + const char* origin_id, + const char* timestamp, + const char* raw, + char* buffer, + size_t buffer_size +) { + JsonDocument doc; + JsonObject root = doc.to(); + + root["origin"] = origin; + root["origin_id"] = origin_id; + root["timestamp"] = timestamp; + root["type"] = "RAW"; + root["data"] = raw; + + return serializeComplete(root, buffer, buffer_size); +} diff --git a/src/helpers/MQTTPayloadBuilder.h b/src/helpers/MQTTPayloadBuilder.h new file mode 100644 index 00000000..57793b13 --- /dev/null +++ b/src/helpers/MQTTPayloadBuilder.h @@ -0,0 +1,72 @@ +#pragma once + +#include +#include +#include + +// Mesh-independent JSON serialization core for MQTT publication payloads. +// MQTTMessageBuilder keeps the firmware-facing API and delegates these three +// deterministic contracts here so they can be exercised by native tests. +class MQTTPayloadBuilder { +public: + static int buildStatusMessage( + JsonDocument& doc, + const char* origin, + const char* origin_id, + const char* model, + const char* firmware_version, + const char* radio, + const char* client_version, + const char* status, + const char* timestamp, + char* buffer, + size_t buffer_size, + int battery_mv = -1, + int uptime_secs = -1, + int errors = -1, + int queue_len = -1, + int noise_floor = -999, + int tx_air_secs = -1, + int rx_air_secs = -1, + int recv_errors = -1, + int internal_heap = -1, + int packets_sent = -1, + int packets_received = -1, + const char* repeat = nullptr + ); + + static int buildPacketMessage( + JsonDocument& doc, + const char* origin, + const char* origin_id, + const char* timestamp, + const char* direction, + const char* time, + const char* date, + int len, + int packet_type, + const char* route, + int payload_len, + const char* raw, + float snr, + int rssi, + float score, + const char* hash, + const uint8_t* path_bytes, + int path_hop_count, + int path_hash_size, + size_t max_path_bytes, + char* buffer, + size_t buffer_size + ); + + static int buildRawMessage( + const char* origin, + const char* origin_id, + const char* timestamp, + const char* raw, + char* buffer, + size_t buffer_size + ); +}; + diff --git a/src/helpers/MQTTTopicRouter.h b/src/helpers/MQTTTopicRouter.h new file mode 100644 index 00000000..db7f5cc2 --- /dev/null +++ b/src/helpers/MQTTTopicRouter.h @@ -0,0 +1,85 @@ +#pragma once + +#include +#include +#include + +#include "MQTTObserverValidation.h" +#include "MQTTTopicTemplate.h" + +// Pure MQTT publication-topic policy shared by MQTTBridge and the native tests. +// Keep these values aligned with MQTTBridge::MQTTMessageType; the bridge passes +// its enum value as an int so this helper stays independent of ESP/Arduino types. +enum MQTTPublicationType { + MQTT_PUBLICATION_STATUS = 0, + MQTT_PUBLICATION_PACKETS = 1, + MQTT_PUBLICATION_RAW = 2, +}; + +enum MQTTTopicRouteStyle { + MQTT_ROUTE_MESHCORE, + MQTT_ROUTE_MESHRANK, + MQTT_ROUTE_CUSTOM, +}; + +static inline bool mqttTopicSlotIndexValid(int index, size_t slot_count) { + return index >= 0 && (size_t)index < slot_count; +} + +static inline const char* mqttPublicationTypeName(int type) { + switch (type) { + case MQTT_PUBLICATION_STATUS: return "status"; + case MQTT_PUBLICATION_PACKETS: return "packets"; + case MQTT_PUBLICATION_RAW: return "raw"; + default: return NULL; + } +} + +static inline bool mqttWriteTopic(char* buf, size_t buf_size, const char* format, + const char* first, const char* second, + const char* third) { + if (!buf || buf_size == 0 || !format || !first || !second || !third) return false; + buf[0] = '\0'; + int written = snprintf(buf, buf_size, format, first, second, third); + return written > 0 && (size_t)written < buf_size; +} + +// Build the complete topic for one publication. MeshRank is deliberately +// packets-only; status and raw are unsupported by the current broker contract. +// 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, + const char* custom_template, + const char* iata, const char* device, + const char* token, + char* buf, size_t buf_size) { + if (!buf || buf_size == 0) return false; + buf[0] = '\0'; + + const char* type_name = mqttPublicationTypeName(type); + if (!type_name) return false; + + switch (style) { + case MQTT_ROUTE_MESHCORE: + if (!mqttIataValid(iata) || strcmp(iata, "XXX") == 0 || !device || device[0] == '\0') { + return false; + } + return mqttWriteTopic(buf, buf_size, "meshcore/%s/%s/%s", iata, device, type_name); + + case MQTT_ROUTE_MESHRANK: + if (type != MQTT_PUBLICATION_PACKETS || !token || token[0] == '\0' || + !device || device[0] == '\0') { + return false; + } + return mqttWriteTopic(buf, buf_size, "meshrank/uplink/%s/%s/%s", + token, device, type_name); + + case MQTT_ROUTE_CUSTOM: + return mqttSubstituteTopic(custom_template, iata, device, token, type_name, + buf, buf_size); + + default: + return false; + } +} + diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 9599d2f2..fb62b79f 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -1,6 +1,8 @@ #include "MQTTBridge.h" +#include "../MQTTConnectionPolicy.h" #include "../MQTTMessageBuilder.h" -#include "../MQTTTopicTemplate.h" +#include "../MQTTPacketQueuePolicy.h" +#include "../MQTTTopicRouter.h" #include "../TxtDataHelpers.h" #include #include @@ -1520,8 +1522,8 @@ void MQTTBridge::maintainSlotConnections() { // JWT tokens require valid timestamps unsigned long clock_sec = current_time; - bool clock_looks_set = (clock_sec >= 1735689600); // 2025-01-01 00:00:00 UTC - bool can_do_jwt = _ntp_synced || clock_looks_set; + bool can_do_jwt = MQTTConnectionPolicy::jwtClockAvailable( + _ntp_synced, static_cast(clock_sec)); // Count connected slots to inform reconnect decisions int connected_count = 0; @@ -1534,8 +1536,8 @@ void MQTTBridge::maintainSlotConnections() { // Time-based guard: block reconnects if any slot reconnected within the last 15 s, // ensuring the previous TLS handshake (and its Core-0-expensive completion events) // finish before the next slot begins its own handshake. - const unsigned long RECONNECT_GUARD_MS = 15000UL; - bool reconnect_attempted_this_cycle = (now_millis - _last_slot_reconnect_ms < RECONNECT_GUARD_MS); + bool reconnect_attempted_this_cycle = MQTTConnectionPolicy::reconnectGuardActive( + static_cast(now_millis), static_cast(_last_slot_reconnect_ms)); // Only allow one full teardown+setup per cycle to limit heap fragmentation // when multiple slots fail simultaneously bool teardown_attempted_this_cycle = false; @@ -1564,11 +1566,10 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns // (worst case the 300 s rung / 30-minute breaker probes) instead of // hammering full TLS handshakes at the 10 s rung — see the onConnect // handler in initSlotClients() for why this doesn't happen on CONNACK. - static const unsigned long BACKOFF_STABLE_RESET_MS = 120000UL; if (slot.connected && (slot.reconnect_backoff != 0 || slot.max_backoff_failures != 0) && - slot.connected_at_ms != 0 && - (now_millis - slot.connected_at_ms) >= BACKOFF_STABLE_RESET_MS) { + MQTTConnectionPolicy::stableConnection(static_cast(now_millis), + static_cast(slot.connected_at_ms))) { MQTT_DEBUG_PRINTLN("MQTT%d stable for %lus - clearing reconnect backoff (was level %d)", index + 1, (now_millis - slot.connected_at_ms) / 1000UL, slot.reconnect_backoff); slot.reconnect_backoff = 0; @@ -1580,23 +1581,19 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns (!slot.preset && slot.audience[0] != '\0'); if (slot_uses_jwt) { // Renew (and below, reconnect) this many seconds before the token's exp - // claim. Scaled to the slot's token lifetime — see tokenRenewalBufferSecs + // claim. Scaled to the slot's token lifetime — see renewalBufferSecs() // for why a flat 60 s lost the renewal race against brokers that enforce // exp on live sessions (waev's 55-minute tokens). - const unsigned long renewal_buffer = tokenRenewalBufferSecs(slotTokenLifetime(index)); - bool token_needs_renewal = false; - if (!time_synced) { - token_needs_renewal = (slot.token_expires_at == 0); - } else { - token_needs_renewal = (slot.token_expires_at == 0) || - !(slot.token_expires_at >= 1000000000) || - (current_time >= slot.token_expires_at) || - (current_time >= (slot.token_expires_at - renewal_buffer)); - } + const unsigned long renewal_buffer = MQTTConnectionPolicy::renewalBufferSecs( + static_cast(slotTokenLifetime(index))); + bool token_needs_renewal = MQTTConnectionPolicy::tokenNeedsRenewal( + time_synced, static_cast(current_time), + static_cast(slot.token_expires_at), + static_cast(renewal_buffer)); // Throttle renewal attempts to once per minute - const unsigned long RENEWAL_THROTTLE_MS = 60000; - bool can_attempt_renewal = (now_millis - slot.last_token_renewal) >= RENEWAL_THROTTLE_MS; + bool can_attempt_renewal = MQTTConnectionPolicy::renewalAttemptAllowed( + static_cast(now_millis), static_cast(slot.last_token_renewal)); if (token_needs_renewal && can_attempt_renewal) { slot.last_token_renewal = now_millis; @@ -1653,11 +1650,10 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns // Periodic probe for circuit-breaker-tripped slots (recovery from transient outages) // Attempts a single reconnect every 30 minutes to see if the server has come back if (slot.circuit_breaker_tripped && !reconnect_attempted) { - static const unsigned long CIRCUIT_BREAKER_PROBE_INTERVAL_MS = 1800000UL; // 30 minutes - unsigned long probe_elapsed = (now_millis >= slot.last_reconnect_attempt) ? - (now_millis - slot.last_reconnect_attempt) : - (ULONG_MAX - slot.last_reconnect_attempt + now_millis + 1); - if (probe_elapsed >= CIRCUIT_BREAKER_PROBE_INTERVAL_MS) { + unsigned long probe_elapsed = MQTTConnectionPolicy::elapsedMs( + static_cast(now_millis), static_cast(slot.last_reconnect_attempt)); + if (MQTTConnectionPolicy::circuitBreakerProbeDue( + static_cast(now_millis), static_cast(slot.last_reconnect_attempt))) { slot.last_reconnect_attempt = now_millis; reconnect_attempted = true; _last_slot_reconnect_ms = now_millis; @@ -1686,24 +1682,18 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns // Reconnect with exponential backoff (for disconnected slots that already have valid config) // Only one reconnect per maintenance cycle to prevent TLS handshakes from blocking other slots if (!slot.connected && slot.initial_connect_done && !slot.circuit_breaker_tripped && !reconnect_attempted) { - static const unsigned long SLOT_BACKOFF_MS[] = { 10000, 30000, 60000, 120000, 300000 }; - static const uint8_t MAX_FAILURES_AT_MAX_BACKOFF = 3; // ~15 min at max backoff before giving up - unsigned long reconnect_elapsed = (now_millis >= slot.last_reconnect_attempt) ? - (now_millis - slot.last_reconnect_attempt) : - (ULONG_MAX - slot.last_reconnect_attempt + now_millis + 1); - unsigned int idx = (slot.reconnect_backoff < 5) ? slot.reconnect_backoff : 4; - unsigned long delay_ms = SLOT_BACKOFF_MS[idx] + (index * 3000UL); // stagger by slot index - if (reconnect_elapsed >= delay_ms) { + if (MQTTConnectionPolicy::reconnectDue( + static_cast(now_millis), static_cast(slot.last_reconnect_attempt), + slot.reconnect_backoff, static_cast(index))) { slot.last_reconnect_attempt = now_millis; - if (slot.reconnect_backoff < 5) { - slot.reconnect_backoff++; - } else { - slot.max_backoff_failures++; - if (slot.max_backoff_failures >= MAX_FAILURES_AT_MAX_BACKOFF) { - slot.circuit_breaker_tripped = true; - MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker tripped after %d failures at max backoff - stopping reconnect attempts. Reconfigure slot to retry.", index + 1, slot.max_backoff_failures); - return; - } + MQTTConnectionPolicy::BackoffAdvance advance = MQTTConnectionPolicy::advanceBackoff( + slot.reconnect_backoff, slot.max_backoff_failures); + slot.reconnect_backoff = advance.reconnect_backoff; + slot.max_backoff_failures = advance.max_backoff_failures; + slot.circuit_breaker_tripped = advance.circuit_breaker_tripped; + if (!advance.should_reconnect) { + MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker tripped after %d failures at max backoff - stopping reconnect attempts. Reconfigure slot to retry.", index + 1, slot.max_backoff_failures); + return; } MQTT_DEBUG_PRINTLN("MQTT%d reconnecting (backoff level %d, failures at max: %d, int_heap=%d)", index + 1, slot.reconnect_backoff, slot.max_backoff_failures, (int)heap_caps_get_free_size(MALLOC_CAP_INTERNAL)); @@ -1740,14 +1730,12 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns // renewal scheduling in maintainSlotConnection() can be derived from it. unsigned long MQTTBridge::slotTokenLifetime(int index) const { const MQTTSlot& slot = _slots[index]; - unsigned long base_lifetime = 86400; // default 24h + unsigned long base_lifetime = MQTTConnectionPolicy::kDefaultJwtLifetimeSecs; if (slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT && slot.preset->token_lifetime > 0) { base_lifetime = slot.preset->token_lifetime; } - // Stagger token expiry per slot to avoid simultaneous renewal/reconnect. - // Use 5% of lifetime per slot, capped at 300s, so short-lived tokens aren't over-reduced. - unsigned long stagger = index * min((unsigned long)300, base_lifetime / 20); - return base_lifetime - stagger; + return MQTTConnectionPolicy::jwtLifetimeSecs( + static_cast(base_lifetime), static_cast(index)); } // How early (seconds before the token's exp claim) to renew the token AND @@ -1763,13 +1751,6 @@ unsigned long MQTTBridge::slotTokenLifetime(int index) const { // lifetime/10 with a 60 s floor and 300 s cap: 24 h tokens renew 5 min early // (unchanged in practice), waev renews ~5 min early with ~5 throttled retry // windows, and degenerate short lifetimes still renew inside their validity. -unsigned long MQTTBridge::tokenRenewalBufferSecs(unsigned long lifetime_secs) { - unsigned long buffer = lifetime_secs / 10; - if (buffer < 60) buffer = 60; - if (buffer > 300) buffer = 300; - return buffer; -} - bool MQTTBridge::createSlotAuthToken(int index) { if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; MQTTSlot& slot = _slots[index]; @@ -1885,31 +1866,28 @@ bool MQTTBridge::publishToAllSlots(const char* topic, const char* payload, bool // Presets use hardcoded topic logic; custom slots support user-defined templates. // --------------------------------------------------------------------------- bool MQTTBridge::substituteTopicTemplate(const char* tmpl, MQTTMessageType type, int slot_index, char* buf, size_t buf_size) { - const char* type_str = (type == MSG_STATUS) ? "status" : (type == MSG_PACKETS) ? "packets" : "raw"; - // Pure expansion lives in helpers/MQTTTopicTemplate.h (host-tested). - return mqttSubstituteTopic(tmpl, _iata, _device_id, - _obs->mqtt_slot_token[slot_index], type_str, buf, buf_size); + return mqttBuildPublicationTopic(MQTT_ROUTE_CUSTOM, (int)type, tmpl, + _iata, _device_id, _obs->mqtt_slot_token[slot_index], + buf, buf_size); } bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_buf, size_t buf_size) { - if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false; + static_assert( + static_cast(MSG_STATUS) == MQTT_PUBLICATION_STATUS && + static_cast(MSG_PACKETS) == MQTT_PUBLICATION_PACKETS && + static_cast(MSG_RAW) == MQTT_PUBLICATION_RAW, + "topic router enum drift"); + + if (!mqttTopicSlotIndexValid(index, RUNTIME_MQTT_SLOTS)) return false; const MQTTSlot& slot = _slots[index]; // Preset slots: use hardcoded topic logic if (slot.preset) { - if (slot.preset->topic_style == MQTT_TOPIC_MESHRANK) { - // MeshRank: packets only, uses per-slot token in topic path - if (type != MSG_PACKETS) return false; - const char* token = _obs->mqtt_slot_token[index]; - if (!token || token[0] == '\0') return false; - snprintf(topic_buf, buf_size, "meshrank/uplink/%s/%s/packets", token, _device_id); - return true; - } - // MQTT_TOPIC_MESHCORE (default for all other presets) - if (!isIATAValid()) return false; - const char* type_str = (type == MSG_STATUS) ? "status" : (type == MSG_PACKETS) ? "packets" : "raw"; - snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id, type_str); - return true; + MQTTTopicRouteStyle style = (slot.preset->topic_style == MQTT_TOPIC_MESHRANK) + ? MQTT_ROUTE_MESHRANK : MQTT_ROUTE_MESHCORE; + return mqttBuildPublicationTopic(style, (int)type, nullptr, + _iata, _device_id, _obs->mqtt_slot_token[index], + topic_buf, buf_size); } // Custom slots: use topic template if set, otherwise default meshcore format @@ -1917,10 +1895,9 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_ return substituteTopicTemplate(_obs->mqtt_slot_topic[index], type, index, topic_buf, buf_size); } // Default: meshcore format - if (!isIATAValid()) return false; - const char* type_str = (type == MSG_STATUS) ? "status" : (type == MSG_PACKETS) ? "packets" : "raw"; - snprintf(topic_buf, buf_size, "meshcore/%s/%s/%s", _iata, _device_id, type_str); - return true; + return mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, (int)type, nullptr, + _iata, _device_id, _obs->mqtt_slot_token[index], + topic_buf, buf_size); } void MQTTBridge::publishStatusToSlot(int index) { @@ -2451,7 +2428,9 @@ void MQTTBridge::processPacketQueue() { // Flush stale packets after extended disconnect if (_queue_disconnected_since == 0) { _queue_disconnected_since = now; - } else if ((now - _queue_disconnected_since) >= QUEUE_STALE_MS) { + } else if (MQTTPacketQueuePolicy::shouldFlushDisconnected( + static_cast(now), + static_cast(_queue_disconnected_since))) { QueuedPacket discard; while (xQueueReceive(_packet_queue_handle, &discard, 0) == pdTRUE) {} _queue_count = 0; @@ -2467,19 +2446,18 @@ void MQTTBridge::processPacketQueue() { // Adaptive drain: burst-process when queue has backlog, gentle otherwise int processed = 0; - int max_per_loop = (_queue_count > 5) ? 5 : 1; + const MQTTPacketQueuePolicy::DrainBudget drain_budget = + MQTTPacketQueuePolicy::drainBudget(static_cast(_queue_count)); unsigned long loop_start_time = millis(); - const unsigned long MAX_PROCESSING_TIME_MS = (_queue_count > 5) ? 100 : 30; - static const uint8_t MAX_QOS0_RETRY_ATTEMPTS = 3; - static const unsigned long RETRY_DELAY_BASE_MS = 300UL; - static const unsigned long RETRY_DELAY_JITTER_MS = 200UL; #ifdef MQTT_DIAG_VERBOSE static unsigned long last_retry_schedule_log = 0; #endif - while (processed < max_per_loop) { - unsigned long elapsed = millis() - loop_start_time; - if (elapsed > MAX_PROCESSING_TIME_MS) { + while (processed < drain_budget.max_packets) { + if (!MQTTPacketQueuePolicy::drainTimeAvailable( + static_cast(millis()), + static_cast(loop_start_time), + drain_budget.max_time_ms)) { break; } @@ -2490,7 +2468,10 @@ void MQTTBridge::processPacketQueue() { } unsigned long now_ms = millis(); - if (queued.next_retry_ms != 0 && now_ms < queued.next_retry_ms) { + if (!MQTTPacketQueuePolicy::retryReady( + static_cast(now_ms), + static_cast(queued.next_retry_ms), + queued.retry_attempts)) { // Not ready yet; put it back and stop draining this cycle. xQueueSend(_packet_queue_handle, &queued, 0); break; @@ -2520,16 +2501,22 @@ void MQTTBridge::processPacketQueue() { } bool any_published = packet_published || raw_published; - if (!any_published && queued.retry_attempts < MAX_QOS0_RETRY_ATTEMPTS) { - queued.retry_attempts++; - unsigned long retry_delay_ms = RETRY_DELAY_BASE_MS + (now_ms % RETRY_DELAY_JITTER_MS); - queued.next_retry_ms = now_ms + retry_delay_ms; + const MQTTPacketQueuePolicy::RetryDecision retry = + MQTTPacketQueuePolicy::retryDecision( + any_published, queued.retry_attempts, + static_cast(now_ms)); + if (retry.action == MQTTPacketQueuePolicy::RetryAction::Schedule) { + queued.retry_attempts = retry.retry_attempts; + queued.next_retry_ms = retry.next_retry_ms; #ifdef MQTT_DIAG_VERBOSE if (now_ms - last_retry_schedule_log > 5000UL) { - unsigned long age_ms = (queued.timestamp > 0 && now_ms >= queued.timestamp) ? (now_ms - queued.timestamp) : 0; + unsigned long age_ms = queued.timestamp > 0 + ? MQTTPacketQueuePolicy::elapsedMs(static_cast(now_ms), + static_cast(queued.timestamp)) + : 0; MQTT_DEBUG_PRINTLN("Retry scheduled: attempt=%u/%u delay=%lu age=%lu q=%u pkt_type=%u packet_ok=%d raw_ok=%d", - (unsigned)queued.retry_attempts, (unsigned)MAX_QOS0_RETRY_ATTEMPTS, - retry_delay_ms, age_ms, (unsigned)uxQueueMessagesWaiting(_packet_queue_handle), + (unsigned)queued.retry_attempts, (unsigned)MQTTPacketQueuePolicy::kMaxQos0RetryAttempts, + (unsigned long)retry.delay_ms, age_ms, (unsigned)uxQueueMessagesWaiting(_packet_queue_handle), (unsigned)queued.packet_copy.getPayloadType(), packet_published ? 1 : 0, raw_published ? 1 : 0); last_retry_schedule_log = now_ms; } @@ -2537,13 +2524,16 @@ void MQTTBridge::processPacketQueue() { if (xQueueSend(_packet_queue_handle, &queued, 0) != pdTRUE) { MQTT_DEBUG_PRINTLN("Retry requeue failed, dropping packet (attempt=%u)", queued.retry_attempts); } - } else if (!any_published) { + } else if (retry.action == MQTTPacketQueuePolicy::RetryAction::Drop) { // Intentional: QoS0 best-effort packets are dropped silently in normal // builds; detailed exhaustion logs are only emitted in verbose mode. #ifdef MQTT_DIAG_VERBOSE static unsigned long last_retry_drop_log = 0; if (now_ms - last_retry_drop_log > 60000UL) { - unsigned long age_ms = (queued.timestamp > 0 && now_ms >= queued.timestamp) ? (now_ms - queued.timestamp) : 0; + unsigned long age_ms = queued.timestamp > 0 + ? MQTTPacketQueuePolicy::elapsedMs(static_cast(now_ms), + static_cast(queued.timestamp)) + : 0; MQTT_DEBUG_PRINTLN("Packet dropped after retry exhaustion (attempts=%u age=%lu pkt_type=%u packet_ok=%d raw_ok=%d)", queued.retry_attempts, age_ms, (unsigned)queued.packet_copy.getPayloadType(), packet_published ? 1 : 0, raw_published ? 1 : 0); @@ -2558,6 +2548,7 @@ void MQTTBridge::processPacketQueue() { #else // Non-ESP32: Use circular buffer if (_queue_count == 0) { + _queue_disconnected_since = 0; return; } @@ -2570,33 +2561,48 @@ void MQTTBridge::processPacketQueue() { MQTT_DEBUG_PRINTLN("Queue has %d packets but no slots connected", _queue_count); _last_no_broker_log = now; } + if (_queue_disconnected_since == 0) { + _queue_disconnected_since = now; + } else if (MQTTPacketQueuePolicy::shouldFlushDisconnected( + static_cast(now), + static_cast(_queue_disconnected_since))) { + while (_queue_count > 0) { + dequeuePacket(); + } + MQTT_DEBUG_PRINTLN("Flushed stale packet queue after %lu ms disconnected", + now - _queue_disconnected_since); + _queue_disconnected_since = now; + } } return; } + _queue_disconnected_since = 0; _last_no_broker_log = 0; // Adaptive drain: burst-process when queue has backlog, gentle otherwise int processed = 0; - int max_per_loop = (_queue_count > 5) ? 5 : 1; + const MQTTPacketQueuePolicy::DrainBudget drain_budget = + MQTTPacketQueuePolicy::drainBudget(static_cast(_queue_count)); unsigned long loop_start_time = millis(); - const unsigned long MAX_PROCESSING_TIME_MS = (_queue_count > 5) ? 100 : 30; - static const uint8_t MAX_QOS0_RETRY_ATTEMPTS = 3; - static const unsigned long RETRY_DELAY_BASE_MS = 300UL; - static const unsigned long RETRY_DELAY_JITTER_MS = 200UL; #ifdef MQTT_DIAG_VERBOSE static unsigned long last_retry_schedule_log = 0; #endif - while (_queue_count > 0 && processed < max_per_loop) { - unsigned long elapsed = millis() - loop_start_time; - if (elapsed > MAX_PROCESSING_TIME_MS) { + while (_queue_count > 0 && processed < drain_budget.max_packets) { + if (!MQTTPacketQueuePolicy::drainTimeAvailable( + static_cast(millis()), + static_cast(loop_start_time), + drain_budget.max_time_ms)) { break; } QueuedPacket& queued = _packet_queue[_queue_head]; unsigned long now_ms = millis(); - if (queued.next_retry_ms != 0 && now_ms < queued.next_retry_ms) { + if (!MQTTPacketQueuePolicy::retryReady( + static_cast(now_ms), + static_cast(queued.next_retry_ms), + queued.retry_attempts)) { break; } @@ -2621,28 +2627,37 @@ void MQTTBridge::processPacketQueue() { } bool any_published = packet_published || raw_published; - if (!any_published && queued.retry_attempts < MAX_QOS0_RETRY_ATTEMPTS) { - queued.retry_attempts++; - unsigned long retry_delay_ms = RETRY_DELAY_BASE_MS + (now_ms % RETRY_DELAY_JITTER_MS); - queued.next_retry_ms = now_ms + retry_delay_ms; + const MQTTPacketQueuePolicy::RetryDecision retry = + MQTTPacketQueuePolicy::retryDecision( + any_published, queued.retry_attempts, + static_cast(now_ms)); + if (retry.action == MQTTPacketQueuePolicy::RetryAction::Schedule) { + queued.retry_attempts = retry.retry_attempts; + queued.next_retry_ms = retry.next_retry_ms; #ifdef MQTT_DIAG_VERBOSE if (now_ms - last_retry_schedule_log > 5000UL) { - unsigned long age_ms = (queued.timestamp > 0 && now_ms >= queued.timestamp) ? (now_ms - queued.timestamp) : 0; + unsigned long age_ms = queued.timestamp > 0 + ? MQTTPacketQueuePolicy::elapsedMs(static_cast(now_ms), + static_cast(queued.timestamp)) + : 0; MQTT_DEBUG_PRINTLN("Retry scheduled: attempt=%u/%u delay=%lu age=%lu q=%d pkt_type=%u packet_ok=%d raw_ok=%d", - (unsigned)queued.retry_attempts, (unsigned)MAX_QOS0_RETRY_ATTEMPTS, - retry_delay_ms, age_ms, _queue_count, + (unsigned)queued.retry_attempts, (unsigned)MQTTPacketQueuePolicy::kMaxQos0RetryAttempts, + (unsigned long)retry.delay_ms, age_ms, _queue_count, (unsigned)queued.packet_copy.getPayloadType(), packet_published ? 1 : 0, raw_published ? 1 : 0); last_retry_schedule_log = now_ms; } #endif break; // keep packet at head for delayed retry - } else if (!any_published) { + } else if (retry.action == MQTTPacketQueuePolicy::RetryAction::Drop) { // Intentional: QoS0 best-effort packets are dropped silently in normal // builds; detailed exhaustion logs are only emitted in verbose mode. #ifdef MQTT_DIAG_VERBOSE static unsigned long last_retry_drop_log = 0; if (now_ms - last_retry_drop_log > 60000UL) { - unsigned long age_ms = (queued.timestamp > 0 && now_ms >= queued.timestamp) ? (now_ms - queued.timestamp) : 0; + unsigned long age_ms = queued.timestamp > 0 + ? MQTTPacketQueuePolicy::elapsedMs(static_cast(now_ms), + static_cast(queued.timestamp)) + : 0; MQTT_DEBUG_PRINTLN("Packet dropped after retry exhaustion (attempts=%u age=%lu pkt_type=%u packet_ok=%d raw_ok=%d)", queued.retry_attempts, age_ms, (unsigned)queued.packet_copy.getPayloadType(), packet_published ? 1 : 0, raw_published ? 1 : 0); @@ -2970,27 +2985,41 @@ void MQTTBridge::queuePacket(mesh::Packet* packet, bool is_tx) { // Try to send to queue (non-blocking) if (xQueueSend(_packet_queue_handle, &queued, 0) != pdTRUE) { + const MQTTPacketQueuePolicy::EnqueueAction action = + MQTTPacketQueuePolicy::enqueueAction( + static_cast(uxQueueMessagesWaiting(_packet_queue_handle)), + static_cast(MAX_QUEUE_SIZE)); QueuedPacket oldest; - if (xQueueReceive(_packet_queue_handle, &oldest, 0) == pdTRUE) { + if (action == MQTTPacketQueuePolicy::EnqueueAction::EvictOldestThenEnqueue && + xQueueReceive(_packet_queue_handle, &oldest, 0) == pdTRUE) { MQTT_DEBUG_PRINTLN("Queue full, dropping oldest packet reference"); - if (xQueueSend(_packet_queue_handle, &queued, 0) != pdTRUE) { - MQTT_DEBUG_PRINTLN("Failed to queue packet after dropping oldest"); - return; - } - } else { + } else if (action == MQTTPacketQueuePolicy::EnqueueAction::Reject) { + MQTT_DEBUG_PRINTLN("Queue has no capacity"); + return; + } else if (action == MQTTPacketQueuePolicy::EnqueueAction::EvictOldestThenEnqueue) { MQTT_DEBUG_PRINTLN("Queue full and cannot remove oldest packet"); return; } + // If the consumer made room after the failed send, retry without evicting. + if (xQueueSend(_packet_queue_handle, &queued, 0) != pdTRUE) { + MQTT_DEBUG_PRINTLN("Failed to queue packet after overflow handling"); + return; + } } UBaseType_t queue_messages = uxQueueMessagesWaiting(_packet_queue_handle); _queue_count = queue_messages; #else // Non-ESP32: Use circular buffer - if (_queue_count >= MAX_QUEUE_SIZE) { + const MQTTPacketQueuePolicy::EnqueueAction action = + MQTTPacketQueuePolicy::enqueueAction( + static_cast(_queue_count), static_cast(MAX_QUEUE_SIZE)); + if (action == MQTTPacketQueuePolicy::EnqueueAction::EvictOldestThenEnqueue) { QueuedPacket& oldest = _packet_queue[_queue_head]; MQTT_DEBUG_PRINTLN("Queue full, dropping oldest packet (queue size: %d)", _queue_count); dequeuePacket(); + } else if (action == MQTTPacketQueuePolicy::EnqueueAction::Reject) { + return; } QueuedPacket& queued = _packet_queue[_queue_tail]; diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index fee9845d..7e8c0466 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -289,7 +289,6 @@ private: // Queue staleness tracking unsigned long _queue_disconnected_since; // 0 = has connected slots - static const unsigned long QUEUE_STALE_MS = 300000UL; // Flush queue after 5 min disconnected #ifdef WITH_SNMP MeshSNMPAgent* _snmp_agent; @@ -341,7 +340,6 @@ private: void maintainSlotConnection(int index, unsigned long now_millis, unsigned long current_time, bool time_synced, bool& reconnect_attempted, bool& teardown_attempted); bool createSlotAuthToken(int index); // Create/renew JWT token for a slot unsigned long slotTokenLifetime(int index) const; // effective JWT lifetime (preset/default minus slot stagger), seconds - static unsigned long tokenRenewalBufferSecs(unsigned long lifetime_secs); // how early to renew+reconnect before exp bool publishToSlot(int index, const char* topic, const char* payload, bool retained = false, uint8_t qos = 0); bool publishToAllSlots(const char* topic, const char* payload, bool retained = false, uint8_t qos = 0); void publishStatusToSlot(int index); diff --git a/test/README.md b/test/README.md index 8379fc29..1be24625 100644 --- a/test/README.md +++ b/test/README.md @@ -27,6 +27,10 @@ does not reflect the GoogleTest count — run the built binary directly | `test_observer_validation` | `src/helpers/MQTTObserverValidation.h` | IATA (exactly 3 alphanumerics), owner key (64 hex), NTP hostname, and the buffer-fit check behind the #17 length validation — including boundaries and nulls | | `test_webconfig_keys` | `src/helpers/WebConfigKeys.h` | POST-key allowlist, secret detection, slot-index bounds, and the short-key out-of-bounds guard (attacker-supplied keys) | | `test_topic_template` | `src/helpers/MQTTTopicTemplate.h` | `{iata}/{device}/{token}/{type}` expansion, overflow/NUL-termination, and a buffer-size fuzz | +| `test_mqtt_topic_router` | `src/helpers/MQTTTopicRouter.h` | complete preset/custom topic-routing contract; MeshRank packets-only behavior; required identifiers; invalid inputs/slots; exact buffer boundaries | +| `test_mqtt_connection_policy` | `src/helpers/MQTTConnectionPolicy.h` | reconnect guard/backoff/stagger and breaker transitions; stable reset; JWT lifetime/renewal policy; exact timing boundaries and 32-bit `millis()` rollover | +| `test_mqtt_packet_queue_policy` | `src/helpers/MQTTPacketQueuePolicy.h` | queue-full eviction; stale-disconnect flush; adaptive drain limits; bounded QoS0 retries; exact timing boundaries and 32-bit `millis()` rollover | +| `test_mqtt_payload_builder` | `src/helpers/MQTTPayloadBuilder.cpp` | status/packet/raw JSON contracts; optional fields; escaping; RX metrics and path; score handling; exact buffer bounds; maximum representative payloads | | `test_utils` | `src/Utils.cpp` | `Utils::toHex` (upstream) | ## Conventions (and how to add a suite) diff --git a/test/test_mqtt_connection_policy/test_mqtt_connection_policy.cpp b/test/test_mqtt_connection_policy/test_mqtt_connection_policy.cpp new file mode 100644 index 00000000..fd6fbe0b --- /dev/null +++ b/test/test_mqtt_connection_policy/test_mqtt_connection_policy.cpp @@ -0,0 +1,158 @@ +#include +#include +#include + +#include "helpers/MQTTConnectionPolicy.h" + +namespace Policy = MQTTConnectionPolicy; + +TEST(MQTTConnectionPolicy, ElapsedTimeHandlesNormalAndWrappedClocks) { + EXPECT_EQ(4000U, Policy::elapsedMs(5000U, 1000U)); + + const uint32_t before_wrap = std::numeric_limits::max() - 99U; + EXPECT_EQ(150U, Policy::elapsedMs(50U, before_wrap)); +} + +TEST(MQTTConnectionPolicy, CrossSlotReconnectGuardHasExactBoundary) { + EXPECT_TRUE(Policy::reconnectGuardActive(14999U, 0U)); + EXPECT_FALSE(Policy::reconnectGuardActive(15000U, 0U)); + + const uint32_t last = std::numeric_limits::max() - 9999U; + EXPECT_TRUE(Policy::reconnectGuardActive(4999U, last)); + EXPECT_FALSE(Policy::reconnectGuardActive(5000U, last)); +} + +TEST(MQTTConnectionPolicy, StableResetRequiresARealStartAndFullWindow) { + EXPECT_FALSE(Policy::stableConnection(500000U, 0U)); + EXPECT_FALSE(Policy::stableConnection(120999U, 1000U)); + EXPECT_TRUE(Policy::stableConnection(121000U, 1000U)); +} + +TEST(MQTTConnectionPolicy, StableResetWindowSurvivesMillisRollover) { + const uint32_t connected_at = std::numeric_limits::max() - 59999U; + EXPECT_FALSE(Policy::stableConnection(59999U, connected_at)); + EXPECT_TRUE(Policy::stableConnection(60000U, connected_at)); +} + +TEST(MQTTConnectionPolicy, BackoffLadderSaturatesAtFiveMinutes) { + EXPECT_EQ(10000U, Policy::reconnectBackoffMs(0)); + EXPECT_EQ(30000U, Policy::reconnectBackoffMs(1)); + EXPECT_EQ(60000U, Policy::reconnectBackoffMs(2)); + EXPECT_EQ(120000U, Policy::reconnectBackoffMs(3)); + EXPECT_EQ(300000U, Policy::reconnectBackoffMs(4)); + EXPECT_EQ(300000U, Policy::reconnectBackoffMs(5)); + EXPECT_EQ(300000U, Policy::reconnectBackoffMs(255)); +} + +TEST(MQTTConnectionPolicy, LaterSlotsReceiveThreeSecondStagger) { + EXPECT_EQ(10000U, Policy::reconnectDelayMs(0, 0)); + EXPECT_EQ(13000U, Policy::reconnectDelayMs(0, 1)); + EXPECT_EQ(315000U, Policy::reconnectDelayMs(5, 5)); +} + +TEST(MQTTConnectionPolicy, ReconnectDueUsesDelayBoundaryAndWrapSafeElapsedTime) { + EXPECT_FALSE(Policy::reconnectDue(12999U, 0U, 0, 1)); + EXPECT_TRUE(Policy::reconnectDue(13000U, 0U, 0, 1)); + + const uint32_t last = std::numeric_limits::max() - 4999U; + EXPECT_FALSE(Policy::reconnectDue(4999U, last, 0, 0)); + EXPECT_TRUE(Policy::reconnectDue(5000U, last, 0, 0)); +} + +TEST(MQTTConnectionPolicy, BackoffAdvanceClimbsThenCountsFailuresAtMaximum) { + Policy::BackoffAdvance first = Policy::advanceBackoff(0, 0); + EXPECT_EQ(1, first.reconnect_backoff); + EXPECT_EQ(0, first.max_backoff_failures); + EXPECT_FALSE(first.circuit_breaker_tripped); + EXPECT_TRUE(first.should_reconnect); + + Policy::BackoffAdvance enters_maximum = Policy::advanceBackoff(4, 0); + EXPECT_EQ(5, enters_maximum.reconnect_backoff); + EXPECT_EQ(0, enters_maximum.max_backoff_failures); + EXPECT_FALSE(enters_maximum.circuit_breaker_tripped); + EXPECT_TRUE(enters_maximum.should_reconnect); + + Policy::BackoffAdvance first_max_failure = Policy::advanceBackoff(5, 0); + EXPECT_EQ(5, first_max_failure.reconnect_backoff); + EXPECT_EQ(1, first_max_failure.max_backoff_failures); + EXPECT_FALSE(first_max_failure.circuit_breaker_tripped); + EXPECT_TRUE(first_max_failure.should_reconnect); +} + +TEST(MQTTConnectionPolicy, ThirdFailureAtMaximumTripsWithoutAnotherHandshake) { + Policy::BackoffAdvance result = Policy::advanceBackoff(5, 2); + EXPECT_EQ(5, result.reconnect_backoff); + EXPECT_EQ(3, result.max_backoff_failures); + EXPECT_TRUE(result.circuit_breaker_tripped); + EXPECT_FALSE(result.should_reconnect); +} + +TEST(MQTTConnectionPolicy, CircuitBreakerProbeHasExactThirtyMinuteBoundary) { + EXPECT_FALSE(Policy::circuitBreakerProbeDue(1799999U, 0U)); + EXPECT_TRUE(Policy::circuitBreakerProbeDue(1800000U, 0U)); + + const uint32_t last = std::numeric_limits::max() - 899999U; + EXPECT_FALSE(Policy::circuitBreakerProbeDue(899999U, last)); + EXPECT_TRUE(Policy::circuitBreakerProbeDue(900000U, last)); +} + +TEST(MQTTConnectionPolicy, JwtLifetimeUsesCappedPerSlotStagger) { + EXPECT_EQ(86400U, Policy::jwtLifetimeSecs(86400U, 0)); + EXPECT_EQ(86100U, Policy::jwtLifetimeSecs(86400U, 1)); + EXPECT_EQ(84900U, Policy::jwtLifetimeSecs(86400U, 5)); +} + +TEST(MQTTConnectionPolicy, ShortJwtLifetimeUsesFivePercentPerSlot) { + EXPECT_EQ(3300U, Policy::jwtLifetimeSecs(3300U, 0)); + EXPECT_EQ(3135U, Policy::jwtLifetimeSecs(3300U, 1)); + EXPECT_EQ(2970U, Policy::jwtLifetimeSecs(3300U, 2)); + EXPECT_EQ(85U, Policy::jwtLifetimeSecs(100U, 3)); +} + +TEST(MQTTConnectionPolicy, JwtLifetimeCannotUnderflowForUnexpectedSlotInput) { + EXPECT_EQ(0U, Policy::jwtLifetimeSecs(100U, 255)); +} + +TEST(MQTTConnectionPolicy, RenewalBufferHasOneMinuteFloorAndFiveMinuteCap) { + EXPECT_EQ(60U, Policy::renewalBufferSecs(0U)); + EXPECT_EQ(60U, Policy::renewalBufferSecs(599U)); + EXPECT_EQ(60U, Policy::renewalBufferSecs(600U)); + EXPECT_EQ(299U, Policy::renewalBufferSecs(2999U)); + EXPECT_EQ(300U, Policy::renewalBufferSecs(3000U)); + EXPECT_EQ(300U, Policy::renewalBufferSecs(86400U)); +} + +TEST(MQTTConnectionPolicy, UnsynchronizedClockOnlyCreatesAMissingToken) { + EXPECT_TRUE(Policy::tokenNeedsRenewal(false, 0U, 0U, 300U)); + EXPECT_FALSE(Policy::tokenNeedsRenewal(false, 0U, 1735693200U, 300U)); +} + +TEST(MQTTConnectionPolicy, SyncedClockRenewsInvalidExpiredOrImminentTokens) { + const uint32_t expires = 1735693200U; + EXPECT_TRUE(Policy::tokenNeedsRenewal(true, 1735689000U, 0U, 300U)); + EXPECT_TRUE(Policy::tokenNeedsRenewal(true, 1735689000U, 999999999U, 300U)); + EXPECT_FALSE(Policy::tokenNeedsRenewal(true, expires - 301U, expires, 300U)); + EXPECT_TRUE(Policy::tokenNeedsRenewal(true, expires - 300U, expires, 300U)); + EXPECT_TRUE(Policy::tokenNeedsRenewal(true, expires, expires, 300U)); + EXPECT_TRUE(Policy::tokenNeedsRenewal(true, expires + 1U, expires, 300U)); +} + +TEST(MQTTConnectionPolicy, RenewalThrottleHasExactBoundaryAndHandlesRollover) { + EXPECT_FALSE(Policy::renewalAttemptAllowed(59999U, 0U)); + EXPECT_TRUE(Policy::renewalAttemptAllowed(60000U, 0U)); + + const uint32_t last = std::numeric_limits::max() - 29999U; + EXPECT_FALSE(Policy::renewalAttemptAllowed(29999U, last)); + EXPECT_TRUE(Policy::renewalAttemptAllowed(30000U, last)); +} + +TEST(MQTTConnectionPolicy, JwtClockNeedsNtpOrAReasonableWallClock) { + EXPECT_FALSE(Policy::jwtClockAvailable(false, Policy::kJwtClockThreshold - 1U)); + EXPECT_TRUE(Policy::jwtClockAvailable(false, Policy::kJwtClockThreshold)); + EXPECT_TRUE(Policy::jwtClockAvailable(true, 0U)); +} + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/test/test_mqtt_packet_queue_policy/test_mqtt_packet_queue_policy.cpp b/test/test_mqtt_packet_queue_policy/test_mqtt_packet_queue_policy.cpp new file mode 100644 index 00000000..80e4ba26 --- /dev/null +++ b/test/test_mqtt_packet_queue_policy/test_mqtt_packet_queue_policy.cpp @@ -0,0 +1,157 @@ +#include + +#include + +#include "helpers/MQTTPacketQueuePolicy.h" + +namespace QueuePolicy = MQTTPacketQueuePolicy; + +TEST(MQTTPacketQueuePolicy, EnqueuesWhileCapacityRemains) { + EXPECT_EQ(QueuePolicy::EnqueueAction::Enqueue, + QueuePolicy::enqueueAction(0, 6)); + EXPECT_EQ(QueuePolicy::EnqueueAction::Enqueue, + QueuePolicy::enqueueAction(5, 6)); +} + +TEST(MQTTPacketQueuePolicy, EvictsOldestAtOrAboveCapacity) { + EXPECT_EQ(QueuePolicy::EnqueueAction::EvictOldestThenEnqueue, + QueuePolicy::enqueueAction(6, 6)); + EXPECT_EQ(QueuePolicy::EnqueueAction::EvictOldestThenEnqueue, + QueuePolicy::enqueueAction(7, 6)); +} + +TEST(MQTTPacketQueuePolicy, RejectsQueueWithZeroCapacity) { + EXPECT_EQ(QueuePolicy::EnqueueAction::Reject, + QueuePolicy::enqueueAction(0, 0)); +} + +TEST(MQTTPacketQueuePolicy, DisconnectedQueueFlushesAtExactStaleBoundary) { + const uint32_t started = 1000; + EXPECT_FALSE(QueuePolicy::shouldFlushDisconnected( + started + QueuePolicy::kDisconnectedStaleMs - 1, started)); + EXPECT_TRUE(QueuePolicy::shouldFlushDisconnected( + started + QueuePolicy::kDisconnectedStaleMs, started)); + EXPECT_TRUE(QueuePolicy::shouldFlushDisconnected( + started + QueuePolicy::kDisconnectedStaleMs + 1, started)); +} + +TEST(MQTTPacketQueuePolicy, ZeroDisconnectedTimestampMeansNotStarted) { + EXPECT_FALSE(QueuePolicy::shouldFlushDisconnected(UINT32_MAX, 0)); +} + +TEST(MQTTPacketQueuePolicy, DisconnectedStaleTimerSurvivesMillisWrap) { + const uint32_t started = UINT32_MAX - 100; + EXPECT_FALSE(QueuePolicy::shouldFlushDisconnected(198, started, 300)); + EXPECT_TRUE(QueuePolicy::shouldFlushDisconnected(199, started, 300)); +} + +TEST(MQTTPacketQueuePolicy, DrainIsGentleThroughFivePackets) { + for (size_t count = 0; count <= QueuePolicy::kBacklogThreshold; ++count) { + const QueuePolicy::DrainBudget budget = QueuePolicy::drainBudget(count); + EXPECT_EQ(QueuePolicy::kGentleDrainCount, budget.max_packets) << count; + EXPECT_EQ(QueuePolicy::kGentleDrainBudgetMs, budget.max_time_ms) << count; + } +} + +TEST(MQTTPacketQueuePolicy, DrainBurstsAboveFivePackets) { + const QueuePolicy::DrainBudget budget = + QueuePolicy::drainBudget(QueuePolicy::kBacklogThreshold + 1); + EXPECT_EQ(QueuePolicy::kBurstDrainCount, budget.max_packets); + EXPECT_EQ(QueuePolicy::kBurstDrainBudgetMs, budget.max_time_ms); +} + +TEST(MQTTPacketQueuePolicy, DrainTimeBudgetUsesInclusiveBoundary) { + EXPECT_TRUE(QueuePolicy::drainTimeAvailable(129, 100, 30)); + EXPECT_TRUE(QueuePolicy::drainTimeAvailable(130, 100, 30)); + EXPECT_FALSE(QueuePolicy::drainTimeAvailable(131, 100, 30)); +} + +TEST(MQTTPacketQueuePolicy, DrainTimeBudgetSurvivesMillisWrap) { + const uint32_t started = UINT32_MAX - 10; + EXPECT_TRUE(QueuePolicy::drainTimeAvailable(19, started, 30)); + EXPECT_FALSE(QueuePolicy::drainTimeAvailable(20, started, 30)); +} + +TEST(MQTTPacketQueuePolicy, NewPacketIsReadyWithoutRetryDeadline) { + EXPECT_TRUE(QueuePolicy::retryReady(100, 0, 0)); + EXPECT_TRUE(QueuePolicy::retryReady(100, 500, 0)); +} + +TEST(MQTTPacketQueuePolicy, RetryBecomesReadyAtExactDeadline) { + EXPECT_FALSE(QueuePolicy::retryReady(499, 500, 1)); + EXPECT_TRUE(QueuePolicy::retryReady(500, 500, 1)); + EXPECT_TRUE(QueuePolicy::retryReady(501, 500, 1)); +} + +TEST(MQTTPacketQueuePolicy, RetryDeadlineSurvivesMillisWrap) { + const uint32_t deadline = 100; + EXPECT_FALSE(QueuePolicy::retryReady(UINT32_MAX - 50, deadline, 1)); + EXPECT_FALSE(QueuePolicy::retryReady(99, deadline, 1)); + EXPECT_TRUE(QueuePolicy::retryReady(100, deadline, 1)); +} + +TEST(MQTTPacketQueuePolicy, WrappedZeroIsARealRetryDeadline) { + EXPECT_FALSE(QueuePolicy::retryReady(UINT32_MAX, 0, 1)); + EXPECT_TRUE(QueuePolicy::retryReady(0, 0, 1)); +} + +TEST(MQTTPacketQueuePolicy, SuccessfulPublishCompletesWithoutChangingAttempts) { + const QueuePolicy::RetryDecision decision = + QueuePolicy::retryDecision(true, 2, 1234); + EXPECT_EQ(QueuePolicy::RetryAction::Complete, decision.action); + EXPECT_EQ(2, decision.retry_attempts); + EXPECT_EQ(0U, decision.delay_ms); + EXPECT_EQ(0U, decision.next_retry_ms); +} + +TEST(MQTTPacketQueuePolicy, FailedPublishSchedulesBoundedRetry) { + const QueuePolicy::RetryDecision minimum = + QueuePolicy::retryDecision(false, 0, 400); + EXPECT_EQ(QueuePolicy::RetryAction::Schedule, minimum.action); + EXPECT_EQ(1, minimum.retry_attempts); + EXPECT_EQ(300U, minimum.delay_ms); + EXPECT_EQ(700U, minimum.next_retry_ms); + + const QueuePolicy::RetryDecision maximum = + QueuePolicy::retryDecision(false, 1, 599); + EXPECT_EQ(QueuePolicy::RetryAction::Schedule, maximum.action); + EXPECT_EQ(2, maximum.retry_attempts); + EXPECT_EQ(499U, maximum.delay_ms); + EXPECT_EQ(1098U, maximum.next_retry_ms); +} + +TEST(MQTTPacketQueuePolicy, ThirdFailedPublishSchedulesFinalRetry) { + const QueuePolicy::RetryDecision decision = + QueuePolicy::retryDecision(false, 2, 1000); + EXPECT_EQ(QueuePolicy::RetryAction::Schedule, decision.action); + EXPECT_EQ(QueuePolicy::kMaxQos0RetryAttempts, decision.retry_attempts); +} + +TEST(MQTTPacketQueuePolicy, FailureAfterFinalRetryDropsPacket) { + const QueuePolicy::RetryDecision decision = + QueuePolicy::retryDecision(false, QueuePolicy::kMaxQos0RetryAttempts, 1000); + EXPECT_EQ(QueuePolicy::RetryAction::Drop, decision.action); + EXPECT_EQ(QueuePolicy::kMaxQos0RetryAttempts, decision.retry_attempts); + EXPECT_EQ(0U, decision.delay_ms); + EXPECT_EQ(0U, decision.next_retry_ms); +} + +TEST(MQTTPacketQueuePolicy, RetrySchedulingDeadlineMayWrapToZero) { + // This timestamp has jitter 98, so its 398 ms delay wraps to exactly zero. + const uint32_t now = UINT32_MAX - 397; + ASSERT_EQ(98U, now % QueuePolicy::kRetryDelayJitterMs); + const QueuePolicy::RetryDecision decision = + QueuePolicy::retryDecision(false, 0, now); + EXPECT_EQ(QueuePolicy::RetryAction::Schedule, decision.action); + EXPECT_EQ(398U, decision.delay_ms); + EXPECT_EQ(0U, decision.next_retry_ms); + EXPECT_FALSE(QueuePolicy::retryReady(UINT32_MAX, decision.next_retry_ms, + decision.retry_attempts)); + EXPECT_TRUE(QueuePolicy::retryReady(0, decision.next_retry_ms, + decision.retry_attempts)); +} + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp new file mode 100644 index 00000000..7dfcf965 --- /dev/null +++ b/test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp @@ -0,0 +1,236 @@ +#include + +#include +#include +#include +#include +#include + +#include "helpers/MQTTPayloadBuilder.h" + +namespace { + +constexpr const char* kTimestamp = "2026-07-18T12:34:56.123456+00:00"; + +static int buildMinimalStatus(JsonDocument& scratch, char* buffer, size_t buffer_size) { + return MQTTPayloadBuilder::buildStatusMessage( + scratch, "DEN Repeater", "0123456789ABCDEF", "Heltec V3", "v1.16.0", + "915.000000,62.5,7,5", "MeshCore", "online", kTimestamp, + buffer, buffer_size); +} + +static int buildRepresentativePacket(JsonDocument& scratch, const char* direction, + float score, const uint8_t* path, int path_hops, + int path_hash_size, const char* raw, + char* buffer, size_t buffer_size) { + return MQTTPayloadBuilder::buildPacketMessage( + scratch, "DEN Repeater", "0123456789ABCDEF", kTimestamp, direction, + "12:34:56", "18/07/2026", 42, 4, "D", 20, raw, + 10.26f, -87, score, "89ABCDEF01234567", path, path_hops, + path_hash_size, 64, buffer, buffer_size); +} + +TEST(MQTTPayloadBuilder, MinimalStatusHasExactRequiredContract) { + JsonDocument scratch; + char buffer[768]; + int len = buildMinimalStatus(scratch, buffer, sizeof(buffer)); + + ASSERT_GT(len, 0); + EXPECT_EQ(static_cast(len), strlen(buffer)); + EXPECT_STREQ( + "{\"status\":\"online\",\"timestamp\":\"2026-07-18T12:34:56.123456+00:00\"," + "\"origin\":\"DEN Repeater\",\"origin_id\":\"0123456789ABCDEF\"," + "\"model\":\"Heltec V3\",\"firmware_version\":\"v1.16.0\"," + "\"radio\":\"915.000000,62.5,7,5\",\"client_version\":\"MeshCore\"}", + buffer); + + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + EXPECT_FALSE(parsed["repeat"].is()); + EXPECT_FALSE(parsed["stats"].is()); +} + +TEST(MQTTPayloadBuilder, StatusIncludesRepeatAndEveryRequestedStatistic) { + JsonDocument scratch; + char buffer[1024]; + int len = MQTTPayloadBuilder::buildStatusMessage( + scratch, "node", "id", "model", "firmware", "radio", "client", "online", + kTimestamp, buffer, sizeof(buffer), 4200, 86400, 3, 6, -112, + 11, 22, 4, 180864, 31, 47, "on"); + + ASSERT_GT(len, 0); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + EXPECT_STREQ("on", parsed["repeat"].as()); + JsonObject stats = parsed["stats"].as(); + ASSERT_FALSE(stats.isNull()); + EXPECT_EQ(4200, stats["battery_mv"].as()); + EXPECT_EQ(86400, stats["uptime_secs"].as()); + EXPECT_EQ(3, stats["errors"].as()); + EXPECT_EQ(6, stats["queue_len"].as()); + EXPECT_EQ(-112, stats["noise_floor"].as()); + EXPECT_EQ(11, stats["tx_air_secs"].as()); + EXPECT_EQ(22, stats["rx_air_secs"].as()); + EXPECT_EQ(4, stats["recv_errors"].as()); + EXPECT_EQ(180864, stats["internal_heap"].as()); + EXPECT_EQ(31, stats["packets_sent"].as()); + EXPECT_EQ(47, stats["packets_received"].as()); +} + +TEST(MQTTPayloadBuilder, StatusOmissionSentinelsRemainOmitted) { + JsonDocument scratch; + char buffer[768]; + int len = MQTTPayloadBuilder::buildStatusMessage( + scratch, "node", "id", "model", "firmware", "radio", "client", "online", + kTimestamp, buffer, sizeof(buffer), -1, -1, -1, -1, -999, + -1, -1, -1, -1, -1, -1, nullptr); + + ASSERT_GT(len, 0); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + EXPECT_FALSE(parsed["stats"].is()); + EXPECT_FALSE(parsed["repeat"].is()); +} + +TEST(MQTTPayloadBuilder, StringsAreEscapedAndRoundTrip) { + const char* origin = "node \"north\"\\rack\nline"; + const char* model = "Heltec\tV3"; + JsonDocument scratch; + char buffer[1024]; + int len = MQTTPayloadBuilder::buildStatusMessage( + scratch, origin, "id", model, "v1", "radio", "client", "online", + kTimestamp, buffer, sizeof(buffer)); + + ASSERT_GT(len, 0); + EXPECT_NE(std::string::npos, std::string(buffer).find("\\\"north\\\"")); + EXPECT_NE(std::string::npos, std::string(buffer).find("\\\\rack\\nline")); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + EXPECT_STREQ(origin, parsed["origin"].as()); + EXPECT_STREQ(model, parsed["model"].as()); +} + +TEST(MQTTPayloadBuilder, RxPacketIncludesMetricsScaledScoreAndPath) { + const uint8_t path[] = {0xAA, 0xBB, 0x01, 0x2F}; + JsonDocument scratch; + char buffer[2048]; + int len = buildRepresentativePacket( + scratch, "rx", 0.125f, path, 2, 2, "A0B1C2D3", buffer, sizeof(buffer)); + + ASSERT_GT(len, 0); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + EXPECT_STREQ("PACKET", parsed["type"].as()); + EXPECT_STREQ("rx", parsed["direction"].as()); + EXPECT_STREQ("42", parsed["len"].as()); + EXPECT_STREQ("4", parsed["packet_type"].as()); + EXPECT_STREQ("20", parsed["payload_len"].as()); + EXPECT_STREQ("10.3", parsed["SNR"].as()); + EXPECT_STREQ("-87", parsed["RSSI"].as()); + EXPECT_STREQ("125", parsed["score"].as()); + JsonArray parsed_path = parsed["path"].as(); + ASSERT_EQ(2U, parsed_path.size()); + EXPECT_STREQ("aabb", parsed_path[0].as()); + EXPECT_STREQ("012f", parsed_path[1].as()); +} + +TEST(MQTTPayloadBuilder, TxPacketOmitsReceiveOnlyMetricsAndAbsentPath) { + JsonDocument scratch; + char buffer[2048]; + int len = buildRepresentativePacket( + scratch, "tx", 0.5f, nullptr, 0, 0, "A0B1", buffer, sizeof(buffer)); + + ASSERT_GT(len, 0); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + EXPECT_FALSE(parsed["SNR"].is()); + EXPECT_FALSE(parsed["RSSI"].is()); + EXPECT_FALSE(parsed["score"].is()); + EXPECT_FALSE(parsed["path"].is()); +} + +TEST(MQTTPayloadBuilder, RxPacketOmitsUnknownNanScore) { + JsonDocument scratch; + char buffer[2048]; + int len = buildRepresentativePacket( + scratch, "rx", std::nanf(""), nullptr, 0, 0, "A0B1", buffer, sizeof(buffer)); + + ASSERT_GT(len, 0); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + EXPECT_TRUE(parsed["SNR"].is()); + EXPECT_TRUE(parsed["RSSI"].is()); + EXPECT_FALSE(parsed["score"].is()); +} + +TEST(MQTTPayloadBuilder, RawMessageHasExactContractAndEscapesData) { + char buffer[512]; + int len = MQTTPayloadBuilder::buildRawMessage( + "node \"A\"", "id\\1", kTimestamp, "AA\nBB", buffer, sizeof(buffer)); + + ASSERT_GT(len, 0); + EXPECT_EQ(static_cast(len), strlen(buffer)); + EXPECT_STREQ( + "{\"origin\":\"node \\\"A\\\"\",\"origin_id\":\"id\\\\1\"," + "\"timestamp\":\"2026-07-18T12:34:56.123456+00:00\"," + "\"type\":\"RAW\",\"data\":\"AA\\nBB\"}", + buffer); + JsonDocument parsed; + ASSERT_FALSE(deserializeJson(parsed, buffer)); + EXPECT_STREQ("AA\nBB", parsed["data"].as()); +} + +TEST(MQTTPayloadBuilder, ExactOutputSizeSucceedsAndOneByteShortFailsCleanly) { + JsonDocument scratch; + char reference[768]; + int reference_len = buildMinimalStatus(scratch, reference, sizeof(reference)); + ASSERT_GT(reference_len, 0); + + std::vector exact(static_cast(reference_len) + 1); + EXPECT_EQ(reference_len, buildMinimalStatus(scratch, exact.data(), exact.size())); + EXPECT_STREQ(reference, exact.data()); + + std::vector short_buffer(static_cast(reference_len), 'x'); + EXPECT_EQ(0, buildMinimalStatus(scratch, short_buffer.data(), short_buffer.size())); + EXPECT_EQ('\0', short_buffer[0]); + EXPECT_EQ(0, buildMinimalStatus(scratch, short_buffer.data(), 1)); + EXPECT_EQ('\0', short_buffer[0]); + EXPECT_EQ(0, buildMinimalStatus(scratch, nullptr, exact.size())); +} + +TEST(MQTTPayloadBuilder, MaximumRepresentativePacketAndRawPayloadsRemainValid) { + uint8_t path[64]; + for (size_t i = 0; i < sizeof(path); ++i) path[i] = static_cast(i); + std::string raw(510, 'A'); + + JsonDocument scratch; + char packet_buffer[2048]; + int packet_len = buildRepresentativePacket( + scratch, "rx", 1.0f, path, 16, 4, raw.c_str(), + packet_buffer, sizeof(packet_buffer)); + ASSERT_GT(packet_len, 0); + JsonDocument packet; + ASSERT_FALSE(deserializeJson(packet, packet_buffer)); + EXPECT_EQ(510U, strlen(packet["raw"].as())); + JsonArray parsed_path = packet["path"].as(); + ASSERT_EQ(16U, parsed_path.size()); + EXPECT_STREQ("00010203", parsed_path[0].as()); + EXPECT_STREQ("3c3d3e3f", parsed_path[15].as()); + + char raw_buffer[1024]; + int raw_len = MQTTPayloadBuilder::buildRawMessage( + "node", "0123456789ABCDEF", kTimestamp, raw.c_str(), + raw_buffer, sizeof(raw_buffer)); + ASSERT_GT(raw_len, 0); + JsonDocument parsed_raw; + ASSERT_FALSE(deserializeJson(parsed_raw, raw_buffer)); + EXPECT_EQ(510U, strlen(parsed_raw["data"].as())); +} + +} // namespace + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} + diff --git a/test/test_mqtt_topic_router/test_mqtt_topic_router.cpp b/test/test_mqtt_topic_router/test_mqtt_topic_router.cpp new file mode 100644 index 00000000..48881573 --- /dev/null +++ b/test/test_mqtt_topic_router/test_mqtt_topic_router.cpp @@ -0,0 +1,179 @@ +// Host contract tests for the complete MQTT publication-topic routing policy. +#define WITH_MQTT_BRIDGE 1 +#define PROGMEM + +#include +#include +#include + +#include "helpers/MQTTPresets.h" +#include "helpers/MQTTTopicRouter.h" + +namespace { + +constexpr const char* IATA = "DEN"; +constexpr const char* DEVICE = "0123456789ABCDEF"; +constexpr const char* TOKEN = "account-token"; + +struct TypeCase { + int type; + const char* name; +}; + +const TypeCase kTypes[] = { + {MQTT_PUBLICATION_STATUS, "status"}, + {MQTT_PUBLICATION_PACKETS, "packets"}, + {MQTT_PUBLICATION_RAW, "raw"}, +}; + +TEST(MQTTTopicRouter, EveryMeshCorePresetSupportsEveryPublicationType) { + for (int preset_index = 0; preset_index < MQTT_PRESET_COUNT; ++preset_index) { + const MQTTPresetDef& preset = MQTT_PRESETS[preset_index]; + if (preset.topic_style != MQTT_TOPIC_MESHCORE) continue; + + for (const TypeCase& type : kTypes) { + char topic[128]; + ASSERT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, type.type, nullptr, + IATA, DEVICE, TOKEN, topic, sizeof(topic))) + << preset.name << " / " << type.name; + EXPECT_EQ(std::string("meshcore/DEN/0123456789ABCDEF/") + type.name, topic) + << preset.name; + } + } +} + +TEST(MQTTTopicRouter, MeshRankContractIsPacketsOnly) { + const MQTTPresetDef* preset = findMQTTPreset("meshrank"); + ASSERT_NE(nullptr, preset); + ASSERT_EQ(MQTT_TOPIC_MESHRANK, preset->topic_style); + + char topic[128]; + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_STATUS, + nullptr, IATA, DEVICE, TOKEN, topic, sizeof(topic))); + EXPECT_STREQ("", topic); + ASSERT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_PACKETS, + nullptr, IATA, DEVICE, TOKEN, topic, sizeof(topic))); + EXPECT_STREQ("meshrank/uplink/account-token/0123456789ABCDEF/packets", topic); + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_RAW, + nullptr, IATA, DEVICE, TOKEN, topic, sizeof(topic))); + EXPECT_STREQ("", topic); +} + +TEST(MQTTTopicRouter, MeshCoreRequiresUsableIataAndDevice) { + char topic[64]; + const char* invalid_iatas[] = {nullptr, "", "XX", "XXXX", "X/X", "XXX"}; + for (const char* iata : invalid_iatas) { + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS, + nullptr, iata, DEVICE, TOKEN, topic, sizeof(topic))); + EXPECT_STREQ("", topic); + } + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS, + nullptr, IATA, nullptr, TOKEN, topic, sizeof(topic))); + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS, + nullptr, IATA, "", TOKEN, topic, sizeof(topic))); +} + +TEST(MQTTTopicRouter, MeshRankRequiresTokenAndDeviceButNotIata) { + char topic[128]; + const char* missing_tokens[] = {nullptr, ""}; + for (const char* token : missing_tokens) { + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_PACKETS, + nullptr, nullptr, DEVICE, token, topic, sizeof(topic))); + EXPECT_STREQ("", topic); + } + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_PACKETS, + nullptr, nullptr, nullptr, TOKEN, topic, sizeof(topic))); + EXPECT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_PACKETS, + nullptr, nullptr, DEVICE, TOKEN, topic, sizeof(topic))); +} + +TEST(MQTTTopicRouter, CustomTemplateExpandsEveryType) { + for (const TypeCase& type : kTypes) { + char topic[128]; + ASSERT_TRUE(mqttBuildPublicationTopic( + MQTT_ROUTE_CUSTOM, type.type, "custom/{iata}/{token}/{device}/{type}", + IATA, DEVICE, TOKEN, topic, sizeof(topic))); + EXPECT_EQ(std::string("custom/DEN/account-token/0123456789ABCDEF/") + type.name, topic); + } +} + +TEST(MQTTTopicRouter, CustomLiteralDoesNotRequireIataTokenOrDevice) { + char topic[32]; + ASSERT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_CUSTOM, MQTT_PUBLICATION_RAW, + "private/raw", nullptr, nullptr, nullptr, + topic, sizeof(topic))); + EXPECT_STREQ("private/raw", topic); +} + +TEST(MQTTTopicRouter, EmptyCustomTemplateFailsRatherThanFallingBackImplicitly) { + char topic[64]; + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_CUSTOM, MQTT_PUBLICATION_STATUS, + "", IATA, DEVICE, TOKEN, topic, sizeof(topic))); + EXPECT_STREQ("", topic); + + // MQTTBridge selects the MeshCore style explicitly for a custom slot whose + // template is empty; make that default contract visible here. + EXPECT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS, + nullptr, IATA, DEVICE, TOKEN, topic, sizeof(topic))); + EXPECT_STREQ("meshcore/DEN/0123456789ABCDEF/status", topic); +} + +TEST(MQTTTopicRouter, FormattedTopicsRequireRoomForTerminator) { + const char* expected = "meshcore/DEN/0123456789ABCDEF/status"; + const size_t exact_size = strlen(expected) + 1; + char exact[64]; + ASSERT_LE(exact_size, sizeof(exact)); + EXPECT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS, + nullptr, IATA, DEVICE, TOKEN, exact, exact_size)); + EXPECT_STREQ(expected, exact); + + char short_buf[64]; + memset(short_buf, 0x7f, sizeof(short_buf)); + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS, + nullptr, IATA, DEVICE, TOKEN, + short_buf, exact_size - 1)); + EXPECT_EQ('\0', short_buf[exact_size - 2]); +} + +TEST(MQTTTopicRouter, CustomTopicHonorsExactBoundary) { + const char* expected = "custom/DEN/raw"; + char exact[15]; + static_assert(sizeof(exact) == 15, "fixture includes the terminator"); + EXPECT_TRUE(mqttBuildPublicationTopic(MQTT_ROUTE_CUSTOM, MQTT_PUBLICATION_RAW, + "custom/{iata}/{type}", IATA, DEVICE, TOKEN, + exact, sizeof(exact))); + EXPECT_STREQ(expected, exact); + + char short_buf[14]; + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_CUSTOM, MQTT_PUBLICATION_RAW, + "custom/{iata}/{type}", IATA, DEVICE, TOKEN, + short_buf, sizeof(short_buf))); + EXPECT_LT(strlen(short_buf), sizeof(short_buf)); +} + +TEST(MQTTTopicRouter, RejectsInvalidStyleTypeSlotAndOutput) { + char topic[64] = "dirty"; + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, 99, nullptr, + IATA, DEVICE, TOKEN, topic, sizeof(topic))); + EXPECT_STREQ("", topic); + EXPECT_FALSE(mqttBuildPublicationTopic(static_cast(99), + MQTT_PUBLICATION_STATUS, nullptr, + IATA, DEVICE, TOKEN, topic, sizeof(topic))); + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS, + nullptr, IATA, DEVICE, TOKEN, nullptr, 64)); + EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHCORE, MQTT_PUBLICATION_STATUS, + nullptr, IATA, DEVICE, TOKEN, topic, 0)); + + EXPECT_FALSE(mqttTopicSlotIndexValid(-1, RUNTIME_MQTT_SLOTS)); + EXPECT_TRUE(mqttTopicSlotIndexValid(0, RUNTIME_MQTT_SLOTS)); + EXPECT_TRUE(mqttTopicSlotIndexValid(RUNTIME_MQTT_SLOTS - 1, RUNTIME_MQTT_SLOTS)); + EXPECT_FALSE(mqttTopicSlotIndexValid(RUNTIME_MQTT_SLOTS, RUNTIME_MQTT_SLOTS)); + EXPECT_FALSE(mqttTopicSlotIndexValid(0, 0)); +} + +} // namespace + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/variants/heltec_t190/platformio.ini b/variants/heltec_t190/platformio.ini index 5a9e8db2..96c1bb28 100644 --- a/variants/heltec_t190/platformio.ini +++ b/variants/heltec_t190/platformio.ini @@ -128,7 +128,7 @@ lib_deps = ${Heltec_T190_base.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -221,7 +221,7 @@ lib_deps = ${Heltec_T190_base.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 diff --git a/variants/heltec_tracker_v2/platformio.ini b/variants/heltec_tracker_v2/platformio.ini index 512a4543..523b518c 100644 --- a/variants/heltec_tracker_v2/platformio.ini +++ b/variants/heltec_tracker_v2/platformio.ini @@ -200,7 +200,7 @@ lib_deps = ${Heltec_tracker_v1_1.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -236,7 +236,7 @@ lib_deps = ${Heltec_tracker_v2.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -272,7 +272,7 @@ lib_deps = ${Heltec_tracker_v1_1.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -308,7 +308,7 @@ lib_deps = ${Heltec_tracker_v2.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 diff --git a/variants/heltec_v3/platformio.ini b/variants/heltec_v3/platformio.ini index f9a81bb1..541a6c96 100644 --- a/variants/heltec_v3/platformio.ini +++ b/variants/heltec_v3/platformio.ini @@ -153,7 +153,7 @@ lib_deps = ${Heltec_lora32_v3.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone 0neblock/SNMP_Agent @@ -223,7 +223,7 @@ lib_deps = ${Heltec_lora32_v3.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -522,7 +522,7 @@ lib_deps = ${Heltec_lora32_v3.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone 0neblock/SNMP_Agent @@ -562,7 +562,7 @@ lib_deps = ${Heltec_lora32_v3.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 diff --git a/variants/heltec_v4/platformio.ini b/variants/heltec_v4/platformio.ini index f3622bc5..caffe9b4 100644 --- a/variants/heltec_v4/platformio.ini +++ b/variants/heltec_v4/platformio.ini @@ -192,7 +192,7 @@ lib_deps = ${heltec_v4_oled.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -246,7 +246,7 @@ lib_deps = ${heltec_v4_oled.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -336,7 +336,7 @@ lib_deps = ${heltec_v4_oled.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -378,7 +378,7 @@ lib_deps = ${heltec_v4_oled.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 diff --git a/variants/lilygo_t3s3/platformio.ini b/variants/lilygo_t3s3/platformio.ini index 4a259ee7..8c93d6b3 100644 --- a/variants/lilygo_t3s3/platformio.ini +++ b/variants/lilygo_t3s3/platformio.ini @@ -140,7 +140,7 @@ lib_deps = ${LilyGo_T3S3_sx1262.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -180,7 +180,7 @@ lib_deps = ${LilyGo_T3S3_sx1262.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 diff --git a/variants/lilygo_tbeam_1w/platformio.ini b/variants/lilygo_tbeam_1w/platformio.ini index 827cac38..cdcf3126 100644 --- a/variants/lilygo_tbeam_1w/platformio.ini +++ b/variants/lilygo_tbeam_1w/platformio.ini @@ -235,7 +235,7 @@ lib_deps = ${LilyGo_TBeam_1W.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -276,7 +276,7 @@ lib_deps = ${LilyGo_TBeam_1W.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 diff --git a/variants/lilygo_tbeam_SX1262/platformio.ini b/variants/lilygo_tbeam_SX1262/platformio.ini index 2e0d8334..0149e27b 100644 --- a/variants/lilygo_tbeam_SX1262/platformio.ini +++ b/variants/lilygo_tbeam_SX1262/platformio.ini @@ -184,7 +184,7 @@ lib_deps = ${LilyGo_TBeam_SX1262.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -221,7 +221,7 @@ lib_deps = ${LilyGo_TBeam_SX1262.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 diff --git a/variants/lilygo_tbeam_SX1276/platformio.ini b/variants/lilygo_tbeam_SX1276/platformio.ini index 41f56115..160d5ac9 100644 --- a/variants/lilygo_tbeam_SX1276/platformio.ini +++ b/variants/lilygo_tbeam_SX1276/platformio.ini @@ -183,7 +183,7 @@ lib_deps = ${LilyGo_TBeam_SX1276.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -220,7 +220,7 @@ lib_deps = ${LilyGo_TBeam_SX1276.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 diff --git a/variants/lilygo_tbeam_supreme_SX1262/platformio.ini b/variants/lilygo_tbeam_supreme_SX1262/platformio.ini index 7c588e7c..78c18198 100644 --- a/variants/lilygo_tbeam_supreme_SX1262/platformio.ini +++ b/variants/lilygo_tbeam_supreme_SX1262/platformio.ini @@ -143,7 +143,7 @@ lib_deps = ${T_Beam_S3_Supreme_SX1262.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -180,7 +180,7 @@ lib_deps = ${T_Beam_S3_Supreme_SX1262.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 diff --git a/variants/lilygo_tlora_v2_1/platformio.ini b/variants/lilygo_tlora_v2_1/platformio.ini index 37a909f8..28d22605 100644 --- a/variants/lilygo_tlora_v2_1/platformio.ini +++ b/variants/lilygo_tlora_v2_1/platformio.ini @@ -182,7 +182,7 @@ lib_deps = ${LilyGo_TLora_V2_1_1_6_core.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -225,7 +225,7 @@ lib_deps = ${LilyGo_TLora_V2_1_1_6_core.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 diff --git a/variants/rak3112/platformio.ini b/variants/rak3112/platformio.ini index 130b8344..e17550dd 100644 --- a/variants/rak3112/platformio.ini +++ b/variants/rak3112/platformio.ini @@ -139,7 +139,7 @@ lib_deps = ${rak3112.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone 0neblock/SNMP_Agent @@ -195,7 +195,7 @@ lib_deps = ${rak3112.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 diff --git a/variants/station_g2/platformio.ini b/variants/station_g2/platformio.ini index 338f4c31..8a95e992 100644 --- a/variants/station_g2/platformio.ini +++ b/variants/station_g2/platformio.ini @@ -205,7 +205,7 @@ lib_deps = ${Station_G2.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -332,7 +332,7 @@ lib_deps = ${Station_G2.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 diff --git a/variants/xiao_s3_wio/platformio.ini b/variants/xiao_s3_wio/platformio.ini index 8e4314cf..5ed8a4bc 100644 --- a/variants/xiao_s3_wio/platformio.ini +++ b/variants/xiao_s3_wio/platformio.ini @@ -127,7 +127,7 @@ lib_deps = ${Xiao_S3_WIO.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1 @@ -164,7 +164,7 @@ lib_deps = ${Xiao_S3_WIO.lib_deps} ${esp32_ota.lib_deps} elims/PsychicMqttClient@^0.2.4 - bblanchon/ArduinoJson + bblanchon/ArduinoJson @ 7.4.3 arduino-libraries/NTPClient JChristensen/Timezone paulstoffregen/Time@1.6.1