diff --git a/apps/esp32_lvgl/tests/esp32_lvgl_font_hot_path_contract_smoke.cpp b/apps/esp32_lvgl/tests/esp32_lvgl_font_hot_path_contract_smoke.cpp index 78a16b28..1136f735 100644 --- a/apps/esp32_lvgl/tests/esp32_lvgl_font_hot_path_contract_smoke.cpp +++ b/apps/esp32_lvgl/tests/esp32_lvgl_font_hot_path_contract_smoke.cpp @@ -71,6 +71,12 @@ int main(int argc, char** argv) const std::size_t budget_check = position_of(registry, "can_add_content_supplement(pack)"); assert(hot_path_guard < hot_path_check); assert(hot_path_check < budget_check); + const std::size_t deferred_guard = + position_of(registry, "bool can_schedule_deferred_content_supplement_load"); + assert(hot_path_guard < deferred_guard); + assert(contains(registry, "can_load_font_from_activation_path(pack)")); + assert(contains(registry, "bool prepare_content_font_for_text")); + assert(contains(registry, "ScopedExternalFontActivation activation(force_overlay);")); const std::size_t content_ensure = position_of(registry, "bool ensure_content_font_for_text"); const std::size_t supplement_check = position_of( @@ -78,14 +84,24 @@ int main(int argc, char** argv) "!can_activate_content_supplement_for_text(*candidate)"); const std::size_t hot_path_reason = position_of_after(registry, "\"ui_hot_path\"", supplement_check); const std::size_t content_budget_reason = position_of_after(registry, "\"content_budget\"", supplement_check); + const std::size_t deferred_candidate_guard = position_of_after( + registry, + "can_schedule_deferred_content_supplement_load(*candidate)", + supplement_check); const std::size_t deferred_queue = position_of_after( registry, "queue_deferred_content_supplement_load(*candidate, reason);", - supplement_check); + deferred_candidate_guard); + const std::size_t deferred_skip = position_of_after( + registry, + "ui_hot_path_no_deferred_load", + deferred_candidate_guard); assert(content_ensure < supplement_check); assert(supplement_check < hot_path_reason); assert(supplement_check < content_budget_reason); - assert(supplement_check < deferred_queue); + assert(supplement_check < deferred_candidate_guard); + assert(deferred_candidate_guard < deferred_queue); + assert(deferred_candidate_guard < deferred_skip); const std::size_t async_schedule = position_of(registry, "bool schedule_deferred_content_supplement_async"); @@ -123,5 +139,11 @@ int main(int argc, char** argv) "foreground::make_snapshot(foreground::Slot::I18nFontLoad,\n" " foreground::Policy::OverlayImmediate")); + const std::string network_page = read_file( + repo_root / "modules/ui_shared/src/ui/screens/network/network_page_shell.cpp"); + assert(contains(network_page, "prepare_content_font_for_text(g_state.page_body.data(), true)")); + assert(contains(network_page, "::ui::fonts::apply_content_font(label,")); + assert(contains(network_page, "::ui::fonts::apply_content_font(text,")); + return 0; } diff --git a/apps/linux_sim_shell/CMakeLists.txt b/apps/linux_sim_shell/CMakeLists.txt index 776243cc..b54b4dae 100644 --- a/apps/linux_sim_shell/CMakeLists.txt +++ b/apps/linux_sim_shell/CMakeLists.txt @@ -346,6 +346,27 @@ if(BUILD_TESTING) add_test(NAME trailmate_meshtastic_protocol_policy_smoke COMMAND trailmate_meshtastic_protocol_policy_smoke) + add_executable(trailmate_mt_dedup_smoke + "${TRAIL_MATE_REPO_ROOT}/modules/core_chat/tests/test_mt_dedup.cpp" + "${TRAIL_MATE_REPO_ROOT}/modules/core_chat/src/infra/meshtastic/mt_dedup.cpp" + "${TRAIL_MATE_REPO_ROOT}/modules/core_sys/src/sys/clock.cpp") + target_include_directories(trailmate_mt_dedup_smoke + PRIVATE + "${TRAIL_MATE_REPO_ROOT}/modules/core_chat/include" + "${TRAIL_MATE_REPO_ROOT}/modules/core_sys/include") + target_compile_features(trailmate_mt_dedup_smoke + PRIVATE cxx_std_17) + add_test(NAME trailmate_mt_dedup_smoke + COMMAND trailmate_mt_dedup_smoke) + + add_executable(trailmate_meshtastic_mqtt_downlink_contract_smoke + "${TRAIL_MATE_REPO_ROOT}/modules/core_chat/tests/test_meshtastic_mqtt_downlink_contract.cpp") + target_compile_features(trailmate_meshtastic_mqtt_downlink_contract_smoke + PRIVATE cxx_std_17) + add_test(NAME trailmate_meshtastic_mqtt_downlink_contract_smoke + COMMAND trailmate_meshtastic_mqtt_downlink_contract_smoke + "${TRAIL_MATE_REPO_ROOT}") + add_executable(trailmate_protocol_effect_executor_smoke "${TRAIL_MATE_REPO_ROOT}/modules/core_chat/tests/test_protocol_effect_executor.cpp" "${TRAIL_MATE_REPO_ROOT}/modules/core_chat/src/infra/meshtastic/mt_packet_wire.cpp" diff --git a/modules/core_chat/include/chat/infra/meshtastic/mt_dedup.h b/modules/core_chat/include/chat/infra/meshtastic/mt_dedup.h index 279eb70f..2ae94d73 100644 --- a/modules/core_chat/include/chat/infra/meshtastic/mt_dedup.h +++ b/modules/core_chat/include/chat/infra/meshtastic/mt_dedup.h @@ -25,8 +25,8 @@ class MtDedup MtDedup(); ~MtDedup(); - bool isDuplicate(NodeId from_node, uint32_t packet_id); - void markSeen(NodeId from_node, uint32_t packet_id); + bool isDuplicate(NodeId from_node, uint32_t packet_id, uint8_t channel_hash); + void markSeen(NodeId from_node, uint32_t packet_id, uint8_t channel_hash); void cleanup(); private: @@ -34,6 +34,7 @@ class MtDedup { NodeId from; uint32_t id; + uint8_t channel; bool operator<(const PacketKey& other) const { @@ -41,7 +42,11 @@ class MtDedup { return from < other.from; } - return id < other.id; + if (id != other.id) + { + return id < other.id; + } + return channel < other.channel; } }; diff --git a/modules/core_chat/include/chat/runtime/meshtastic_protocol_policy.h b/modules/core_chat/include/chat/runtime/meshtastic_protocol_policy.h index 35d972e2..8597bfa9 100644 --- a/modules/core_chat/include/chat/runtime/meshtastic_protocol_policy.h +++ b/modules/core_chat/include/chat/runtime/meshtastic_protocol_policy.h @@ -155,10 +155,6 @@ inline MeshtasticMqttDownlinkPolicy resolveMeshtasticMqttDownlinkPolicy( policy.reason = MeshtasticMqttDownlinkReason::LocalDestination; return policy; } - - // Zero-hop packets are still transmitted by the gateway; direct neighbors - // receive them, but should not relay them further. Therefore hop_limit is - // not a transmit gate here. policy.transmit_to_mesh = true; policy.reason = MeshtasticMqttDownlinkReason::TransmitToMesh; return policy; diff --git a/modules/core_chat/src/infra/meshtastic/mt_dedup.cpp b/modules/core_chat/src/infra/meshtastic/mt_dedup.cpp index 112cc882..4af83b7d 100644 --- a/modules/core_chat/src/infra/meshtastic/mt_dedup.cpp +++ b/modules/core_chat/src/infra/meshtastic/mt_dedup.cpp @@ -20,18 +20,19 @@ MtDedup::~MtDedup() { } -bool MtDedup::isDuplicate(NodeId from_node, uint32_t packet_id) +bool MtDedup::isDuplicate(NodeId from_node, uint32_t packet_id, uint8_t channel_hash) { cleanup(); PacketKey key; key.from = from_node; key.id = packet_id; + key.channel = channel_hash; return cache_.find(key) != cache_.end(); } -void MtDedup::markSeen(NodeId from_node, uint32_t packet_id) +void MtDedup::markSeen(NodeId from_node, uint32_t packet_id, uint8_t channel_hash) { cleanup(); @@ -44,6 +45,7 @@ void MtDedup::markSeen(NodeId from_node, uint32_t packet_id) PacketKey key; key.from = from_node; key.id = packet_id; + key.channel = channel_hash; PacketEntry entry; entry.timestamp = sys::millis_now(); diff --git a/modules/core_chat/tests/test_meshtastic_mqtt_downlink_contract.cpp b/modules/core_chat/tests/test_meshtastic_mqtt_downlink_contract.cpp new file mode 100644 index 00000000..1ec1430c --- /dev/null +++ b/modules/core_chat/tests/test_meshtastic_mqtt_downlink_contract.cpp @@ -0,0 +1,89 @@ +#include +#include +#include +#include +#include + +namespace +{ + +std::string readFile(const std::filesystem::path& path) +{ + std::ifstream stream(path, std::ios::binary); + assert(stream.is_open()); + std::ostringstream out; + out << stream.rdbuf(); + return out.str(); +} + +bool contains(const std::string& haystack, const char* needle) +{ + return haystack.find(needle) != std::string::npos; +} + +bool notContains(const std::string& haystack, const char* needle) +{ + return !contains(haystack, needle); +} + +std::size_t positionOf(const std::string& haystack, const char* needle) +{ + const auto pos = haystack.find(needle); + assert(pos != std::string::npos); + return pos; +} + +std::size_t positionOfAfter(const std::string& haystack, + const char* needle, + std::size_t offset) +{ + const auto pos = haystack.find(needle, offset); + assert(pos != std::string::npos); + return pos; +} + +} // namespace + +int main(int argc, char** argv) +{ + assert(argc == 2); + const std::filesystem::path repo_root = argv[1]; + const std::string header = readFile( + repo_root / + "platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/meshtastic/mt_adapter.h"); + const std::string source = readFile( + repo_root / + "platform/esp/arduino_common/src/chat/infra/meshtastic/mt_adapter.cpp"); + + assert(contains(header, "struct PendingMqttDownlinkTx")); + assert(contains(header, "struct MqttDownlinkSeenEntry")); + assert(contains(header, "mqtt_downlink_tx_queue_")); + assert(contains(header, "mqtt_downlink_seen_")); + assert(contains(header, "kMqttDownlinkTxDrainPerTick")); + + const std::size_t inject_begin = + positionOf(source, "bool MtAdapter::injectMqttEnvelope"); + const std::size_t handle_begin = + positionOfAfter(source, "bool MtAdapter::handleMqttProxyMessage", inject_begin); + const std::string inject_body = + source.substr(inject_begin, handle_begin - inject_begin); + assert(contains(inject_body, "enqueueMqttDownlinkTx(wire_buffer, wire_size, *tx_header)")); + assert(contains(inject_body, "processReceivedPacket(wire_buffer, wire_size);")); + assert(notContains(inject_body, "transmitWirePacket(wire_buffer, wire_size)")); + assert(positionOf(inject_body, "enqueueMqttDownlinkTx(wire_buffer, wire_size, *tx_header)") < + positionOf(inject_body, "processReceivedPacket(wire_buffer, wire_size);")); + + const std::size_t process_send_begin = + positionOf(source, "void MtAdapter::processSendQueue()"); + const std::size_t process_mqtt_begin = + positionOfAfter(source, "void MtAdapter::processMqttDownlinkTxQueue", process_send_begin); + assert(positionOfAfter(source, "processMqttDownlinkTxQueue(now);", process_send_begin) < + process_mqtt_begin); + assert(contains(source, "drained < kMqttDownlinkTxDrainPerTick")); + assert(contains(source, "isMqttDownlinkRecentlySeen(header.from, header.id, header.channel")); + assert(contains(source, "reason=pending_queue_full")); + assert(contains(source, "reason=airtime_budget")); + assert(contains(source, "\"radio_queue_full\"")); + + return 0; +} diff --git a/modules/core_chat/tests/test_meshtastic_protocol_policy.cpp b/modules/core_chat/tests/test_meshtastic_protocol_policy.cpp index 54a8f32b..fd53fc0a 100644 --- a/modules/core_chat/tests/test_meshtastic_protocol_policy.cpp +++ b/modules/core_chat/tests/test_meshtastic_protocol_policy.cpp @@ -127,6 +127,15 @@ int main() assert(policy.reason == MeshtasticMqttDownlinkReason::TransmitToMesh); } + { + const auto policy = resolveMeshtasticMqttDownlinkPolicy( + "!00000000", 0xA1B3B57CUL, 0x4A59CD8CUL, + 0xB0B0B0B0UL, true); + assert(policy.accept_locally); + assert(policy.transmit_to_mesh); + assert(policy.reason == MeshtasticMqttDownlinkReason::TransmitToMesh); + } + { const auto policy = resolveMeshtasticMqttDownlinkPolicy( "!00000000", 0xA1B3B57CUL, 0x4A59CD8CUL, diff --git a/modules/core_chat/tests/test_mt_dedup.cpp b/modules/core_chat/tests/test_mt_dedup.cpp new file mode 100644 index 00000000..61b68720 --- /dev/null +++ b/modules/core_chat/tests/test_mt_dedup.cpp @@ -0,0 +1,41 @@ +#include "chat/infra/meshtastic/mt_dedup.h" +#include "sys/clock.h" + +#include +#include + +namespace +{ + +uint32_t g_now_ms = 1000; + +uint32_t fake_millis() +{ + return g_now_ms; +} + +} // namespace + +int main() +{ + sys::set_millis_provider(fake_millis); + + chat::meshtastic::MtDedup dedup; + constexpr chat::NodeId from = 0xA1B3B57C; + constexpr uint32_t packet_id = 0x10203040; + + assert(!dedup.isDuplicate(from, packet_id, 0x01)); + dedup.markSeen(from, packet_id, 0x01); + assert(dedup.isDuplicate(from, packet_id, 0x01)); + assert(!dedup.isDuplicate(from, packet_id, 0x02)); + + dedup.markSeen(from, packet_id, 0x02); + assert(dedup.isDuplicate(from, packet_id, 0x01)); + assert(dedup.isDuplicate(from, packet_id, 0x02)); + + g_now_ms += chat::meshtastic::MtDedup::CACHE_TIMEOUT_MS + 30001U; + assert(!dedup.isDuplicate(from, packet_id, 0x01)); + assert(!dedup.isDuplicate(from, packet_id, 0x02)); + + return 0; +} diff --git a/modules/ui_shared/include/ui/localization.h b/modules/ui_shared/include/ui/localization.h index 59aa1810..06a07da6 100644 --- a/modules/ui_shared/include/ui/localization.h +++ b/modules/ui_shared/include/ui/localization.h @@ -44,6 +44,7 @@ const lv_font_t* active_ui_font_fallback(); const lv_font_t* active_content_font_fallback(); const lv_font_t* locale_preview_font(const char* locale_id, const lv_font_t* ascii_font = nullptr); bool ensure_content_font_for_text(const char* text); +bool prepare_content_font_for_text(const char* text, bool force_overlay = true); void log_direct_text_route(const char* owner, const void* label, const char* text); std::size_t ime_count(); const ImeInfo* ime_at(std::size_t index); diff --git a/modules/ui_shared/src/ui/i18n/resource_pack_registry.cpp b/modules/ui_shared/src/ui/i18n/resource_pack_registry.cpp index db21ec77..4620ff29 100644 --- a/modules/ui_shared/src/ui/i18n/resource_pack_registry.cpp +++ b/modules/ui_shared/src/ui/i18n/resource_pack_registry.cpp @@ -89,8 +89,10 @@ constexpr bool kFlashPackStorageEnabled = false; #if defined(ESP_PLATFORM) || defined(ARDUINO_ARCH_ESP32) constexpr bool kAllowSynchronousContentSupplementFontLoad = false; +constexpr bool kAllowDeferredContentSupplementFontLoad = false; #else constexpr bool kAllowSynchronousContentSupplementFontLoad = true; +constexpr bool kAllowDeferredContentSupplementFontLoad = true; #endif enum class FontPackUsage : uint8_t @@ -1670,7 +1672,8 @@ bool can_activate_content_supplement_for_text(const FontPackRecord& pack) // not retried per label. return pack.builtin || is_font_runtime_loaded(pack) || - (can_load_font_from_content_hot_path(pack) && + ((can_load_font_from_content_hot_path(pack) || + can_load_font_from_activation_path(pack)) && font_pack_supports_content(pack) && can_add_content_supplement(pack)); #else @@ -1679,6 +1682,12 @@ bool can_activate_content_supplement_for_text(const FontPackRecord& pack) #endif } +bool can_schedule_deferred_content_supplement_load(const FontPackRecord& pack) +{ + return kAllowDeferredContentSupplementFontLoad || + can_load_font_from_content_hot_path(pack); +} + bool can_preload_small_content_supplement(const FontPackRecord& pack) { if (pack.builtin || is_font_runtime_loaded(pack)) @@ -3444,6 +3453,15 @@ void queue_deferred_content_supplement_load(FontPackRecord& pack, const char* re pack.source_path.empty() ? "" : pack.source_path.c_str()); return; } + if (!can_schedule_deferred_content_supplement_load(pack)) + { + std::printf("%s font load skipped id=%s role=content_supplement reason=ui_hot_path_no_deferred_load active_locale=%s source=%s\n", + kLogTag, + pack.id.c_str(), + s_active_locale ? s_active_locale->id.c_str() : "", + pack.source_path.empty() ? "" : pack.source_path.c_str()); + return; + } if (s_content_supplement_load_async_pending || s_content_supplement_retry_timer) { return; @@ -3582,10 +3600,18 @@ bool ensure_content_font_for_text(const char* text) { log_font_load_deferred(*candidate, "content_supplement", reason); } - else + else if (can_schedule_deferred_content_supplement_load(*candidate)) { queue_deferred_content_supplement_load(*candidate, reason); } + else + { + std::printf("%s font load skipped id=%s role=content_supplement reason=ui_hot_path_no_deferred_load active_locale=%s source=%s\n", + kLogTag, + candidate->id.c_str(), + s_active_locale ? s_active_locale->id.c_str() : "", + candidate->source_path.empty() ? "" : candidate->source_path.c_str()); + } break; } if (!ensure_font_pack_loaded(candidate)) @@ -3642,6 +3668,12 @@ bool ensure_content_font_for_text(const char* text) return missing.empty(); } +bool prepare_content_font_for_text(const char* text, bool force_overlay) +{ + ScopedExternalFontActivation activation(force_overlay); + return ensure_content_font_for_text(text); +} + std::size_t ime_count() { ensure_registry(); diff --git a/modules/ui_shared/src/ui/screens/chat/chat_ui_controller.cpp b/modules/ui_shared/src/ui/screens/chat/chat_ui_controller.cpp index a14b72f7..27790cbb 100644 --- a/modules/ui_shared/src/ui/screens/chat/chat_ui_controller.cpp +++ b/modules/ui_shared/src/ui/screens/chat/chat_ui_controller.cpp @@ -1029,6 +1029,10 @@ void UiController::switchToConversation(chat::ConversationId conv) const ::ui::chat::ConversationId ui_conv = chat_presentation_adapters::toUiConversationId(conv); (void)chat_model_.selectConversation(ui_conv); + if (chat_model_.markRead(ui_conv).ok) + { + conversation_list_dirty_ = true; + } // Update header (prefer contact name, else short_name) std::string title = resolveConversationDisplayName(conv); @@ -1061,10 +1065,6 @@ void UiController::switchToConversation(chat::ConversationId conv) { applySnapshotMessagesToConversation(chat_snapshot_buffer_, *conversation_); } - if (chat_model_.markRead(ui_conv).ok) - { - conversation_list_dirty_ = true; - } } void UiController::switchToCompose(chat::ConversationId conv) diff --git a/modules/ui_shared/src/ui/screens/network/network_page_shell.cpp b/modules/ui_shared/src/ui/screens/network/network_page_shell.cpp index 71e2680c..623fdc45 100644 --- a/modules/ui_shared/src/ui/screens/network/network_page_shell.cpp +++ b/modules/ui_shared/src/ui/screens/network/network_page_shell.cpp @@ -2057,9 +2057,9 @@ void emit_micron_text_token(lv_obj_t* row, lv_obj_set_width(label, LV_SIZE_CONTENT); lv_obj_set_style_max_width(label, LV_PCT(100), LV_PART_MAIN); apply_micron_label_style(label, style, default_bg, font); - ::ui::fonts::apply_localized_font(label, - text, - font ? font : (style.bold ? body_font() : caption_font())); + ::ui::fonts::apply_content_font(label, + text, + font ? font : (style.bold ? body_font() : caption_font())); } void emit_micron_text(lv_obj_t* row, @@ -2282,9 +2282,9 @@ void emit_micron_link(lv_obj_t* row, lv_label_set_long_mode(text, LV_LABEL_LONG_CLIP); lv_obj_set_width(text, LV_SIZE_CONTENT); apply_micron_label_style(text, style, default_bg); - ::ui::fonts::apply_localized_font(text, - visible, - style.bold ? body_font() : caption_font()); + ::ui::fonts::apply_content_font(text, + visible, + style.bold ? body_font() : caption_font()); lv_obj_center(text); if (request_fields) @@ -3601,7 +3601,11 @@ void render_cached_page(const rtpage::Status& status, { clear_viewport(); g_state.rendered_shell_address[0] = '\0'; - render_micron_body(g_state.page_body.data(), body_len, status.truncated); + const std::size_t render_len = + body_len < g_state.page_body.size() ? body_len : g_state.page_body.size() - 1U; + g_state.page_body[render_len] = '\0'; + (void)::ui::i18n::prepare_content_font_for_text(g_state.page_body.data(), true); + render_micron_body(g_state.page_body.data(), render_len, status.truncated); jump_to_rendered_micron_anchor(anchor); } diff --git a/modules/ui_shared/tests/test_chat_presentation_source.cpp b/modules/ui_shared/tests/test_chat_presentation_source.cpp index 99cdc3ef..f3fe55e9 100644 --- a/modules/ui_shared/tests/test_chat_presentation_source.cpp +++ b/modules/ui_shared/tests/test_chat_presentation_source.cpp @@ -471,6 +471,20 @@ const ui::chat::ConversationLocationParticipant* findLocationParticipant( return nullptr; } +uint16_t unreadForConversation(const ui::chat::ChatWorkspaceSnapshot& snapshot, + const ui::chat::ConversationId& id) +{ + for (size_t i = 0; i < snapshot.conversation_count; ++i) + { + if (snapshot.conversations[i].id == id) + { + return snapshot.conversations[i].unread_count; + } + } + assert(false); + return 0; +} + void setNodePosition(::chat::contacts::ContactService& contacts, uint32_t node_id, int32_t lat_e7, @@ -669,6 +683,10 @@ int main() assert(snapshot.messages[0].source_unverified); assert(snapshot.messages[0].sender_node_id == 0x648144D4); assert(std::strcmp(snapshot.messages[0].sender_label.c_str(), "Mother") == 0); + assert(unreadForConversation(snapshot, broadcast) == 1); + assert(sink.markRead(broadcast).ok); + assert(source.buildChatWorkspaceSnapshot(request, snapshot)); + assert(unreadForConversation(snapshot, broadcast) == 0); const uint32_t paging_peer = 0x00ABCDEF; contacts.updateNodeInfo(paging_peer, diff --git a/platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/lxmf/lxmf_adapter.h b/platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/lxmf/lxmf_adapter.h index 8bef6355..d91952b2 100644 --- a/platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/lxmf/lxmf_adapter.h +++ b/platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/lxmf/lxmf_adapter.h @@ -143,6 +143,7 @@ class LxmfAdapter : public IMeshAdapter bool allow_persistence = false; bool allow_peer_projection = false; bool allow_announce_tx = true; + bool allow_propagation_client = false; bool drop_public_discovery = false; const char* phase = "screen"; }; diff --git a/platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/meshtastic/mt_adapter.h b/platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/meshtastic/mt_adapter.h index 040d4d45..a01c0f02 100644 --- a/platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/meshtastic/mt_adapter.h +++ b/platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/meshtastic/mt_adapter.h @@ -225,6 +225,40 @@ class MtAdapter : public chat::IMeshAdapter sys::RingBuffer mqtt_proxy_queue_; MqttProxySettings mqtt_proxy_settings_; + static constexpr std::size_t kMqttDownlinkWireMaxLen = 255; + static constexpr std::size_t kPendingMqttDownlinkTxDepth = 8; + static constexpr std::size_t kMqttDownlinkSeenDepth = 24; + static constexpr uint8_t kMqttDownlinkTxDrainPerTick = 1; + static constexpr uint8_t kMqttDownlinkTxMaxRetries = 2; + static constexpr uint32_t kMqttDownlinkSeenTtlMs = 300000; + + struct PendingMqttDownlinkTx + { + std::array wire{}; + size_t wire_size = 0; + NodeId from = 0; + NodeId to = 0; + MessageId msg_id = 0; + uint8_t channel_hash = 0; + uint8_t retry_count = 0; + uint32_t first_seen_ms = 0; + uint32_t last_attempt_ms = 0; + }; + + struct MqttDownlinkSeenEntry + { + bool used = false; + NodeId from = 0; + MessageId msg_id = 0; + uint8_t channel_hash = 0; + uint32_t seen_ms = 0; + }; + + sys::RingBuffer + mqtt_downlink_tx_queue_; + std::array mqtt_downlink_seen_{}; + size_t mqtt_downlink_seen_next_ = 0; + struct MqttDownlinkScratchBuffers { meshtastic_MeshPacket packet = meshtastic_MeshPacket_init_zero; @@ -424,6 +458,18 @@ class MtAdapter : public chat::IMeshAdapter bool injectMqttEnvelope(const meshtastic_MeshPacket& packet, const char* channel_id, const char* gateway_id); + bool enqueueMqttDownlinkTx(const uint8_t* wire_data, + size_t wire_size, + const PacketHeaderWire& header); + bool isMqttDownlinkRecentlySeen(NodeId from, + MessageId msg_id, + uint8_t channel_hash, + uint32_t now_ms); + void rememberMqttDownlinkSeen(NodeId from, + MessageId msg_id, + uint8_t channel_hash, + uint32_t now_ms); + void processMqttDownlinkTxQueue(uint32_t now_ms); bool queueMqttProxyPublish(const meshtastic_MeshPacket& packet, const char* channel_id); bool queueMqttProxyPublishFromWire(const uint8_t* wire_data, diff --git a/platform/esp/arduino_common/src/chat/infra/lxmf/lxmf_adapter.cpp b/platform/esp/arduino_common/src/chat/infra/lxmf/lxmf_adapter.cpp index b7355f59..3798974c 100644 --- a/platform/esp/arduino_common/src/chat/infra/lxmf/lxmf_adapter.cpp +++ b/platform/esp/arduino_common/src/chat/infra/lxmf/lxmf_adapter.cpp @@ -768,6 +768,40 @@ bool isNomadNetworkNodeAnnounce(const reticulum::ParsedAnnounce& announce) return hashesEqual(expected_name_hash, announce.name_hash, sizeof(expected_name_hash)); } +bool isLxstTelephonyAnnounce(const reticulum::ParsedAnnounce& announce) +{ + if (!announce.valid || !announce.name_hash) + { + return false; + } + + uint8_t expected_name_hash[reticulum::kNameHashSize] = {}; + reticulum::computeNameHash("lxst", "telephony", expected_name_hash); + return hashesEqual(expected_name_hash, announce.name_hash, sizeof(expected_name_hash)); +} + +bool isLxstTelephonyAnnouncePacket(const reticulum::ParsedPacket& packet) +{ + if (!packet.valid || packet.packet_type != reticulum::PacketType::Announce) + { + return false; + } + + reticulum::ParsedAnnounce announce{}; + return reticulum::parseAnnounce(packet, &announce) && + isLxstTelephonyAnnounce(announce); +} + +bool isLoRaInterfaceId(reticulum::interfaces::InterfaceId interface_id) +{ + return interface_id == reticulum::interfaces::kLoRaInterfaceId; +} + +bool isLoRaPath(const runtime::PathEntry* path) +{ + return path && isLoRaInterfaceId(path->interface_id); +} + bool packetContextUsesRawLinkPayload(uint8_t context) { return context == static_cast(reticulum::PacketContext::Keepalive) || @@ -1551,6 +1585,27 @@ void LxmfAdapter::processPropagationClient() } const PropagationPeerState* node = selectActivePropagationPeer(); + const PathEntry* node_path = + node ? path_manager_.findPath(node->propagation_hash, millis(), kPathTtlMs) + : nullptr; + if (node && !node_path) + { + (void)sendPathRequestForDestination(node->propagation_hash); + return; + } + if (node && isLoRaPath(node_path)) + { + if (propagation_client_.hasPendingUploads()) + { + if (PendingPropagationUpload* upload = + propagation_client_.firstPendingUpload()) + { + propagation_client_.markUploadWaitingForNode(*upload); + } + } + return; + } + if (propagation_client_.hasPendingUploads()) { PendingPropagationUpload* pending_upload = @@ -2573,6 +2628,7 @@ bool LxmfAdapter::broadcastSelfIdentity() const bool call_audio_ok = sendAnnounce(LocalDestinationKind::CallAudio); const bool call_audio_complete = !interfaces_.wifiGatewayConfigured() || + !interfaces_.hasReadyWifiGateway() || lastAnnounceTxReachedRequiredInterfaces(call_audio_ok); if (delivery_ok || propagation_ok || call_audio_ok) { @@ -2680,6 +2736,14 @@ MeshActionResult LxmfAdapter::startReticulumAudioCall( const PathEntry* path = path_manager_.findPath(call_destination_hash, millis(), kPathTtlMs); + if (isLoRaPath(path)) + { + char call_hash[12] = {}; + formatHashPrefix(call_destination_hash, call_hash, sizeof(call_hash)); + Serial.printf("[LXMF][CallTX] ignore_path dest=%s reason=lora_not_supported\n", + call_hash); + path = nullptr; + } bool path_requested = false; bool path_waiting = false; if (!path) @@ -3223,6 +3287,7 @@ LxmfAdapter::RuntimeBudget LxmfAdapter::makeRuntimeBudget() const budget.allow_persistence = false; budget.allow_peer_projection = false; budget.allow_announce_tx = false; + budget.allow_propagation_client = false; budget.drop_public_discovery = true; budget.phase = "call"; return budget; @@ -3236,6 +3301,7 @@ LxmfAdapter::RuntimeBudget LxmfAdapter::makeRuntimeBudget() const budget.allow_persistence = false; budget.allow_peer_projection = false; budget.allow_announce_tx = false; + budget.allow_propagation_client = false; budget.drop_public_discovery = true; budget.phase = "nomad"; return budget; @@ -3245,16 +3311,31 @@ LxmfAdapter::RuntimeBudget LxmfAdapter::makeRuntimeBudget() const screen_runtime::is_sleeping() && !screen_runtime::is_saver_active(); if (maintenance_window) { - budget.live_packet_limit = kMaxIngressPacketsPerPoll; + budget.live_packet_limit = 1; budget.deferred_discovery_limit = 1; budget.allow_public_discovery = true; - budget.allow_persistence = true; - budget.allow_peer_projection = true; - budget.allow_announce_tx = true; + budget.allow_persistence = false; + budget.allow_peer_projection = false; + budget.allow_announce_tx = false; + budget.allow_propagation_client = false; budget.phase = "sleep"; return budget; } + if (screen_runtime::is_saver_active()) + { + budget.live_packet_limit = 1; + budget.deferred_discovery_limit = 0; + budget.allow_public_discovery = false; + budget.allow_persistence = false; + budget.allow_peer_projection = false; + budget.allow_announce_tx = false; + budget.allow_propagation_client = false; + budget.drop_public_discovery = true; + budget.phase = "saver"; + return budget; + } + #if defined(TRAIL_MATE_ESP_BOARD_T_DISPLAY_P4) // P4 runs the shared mesh task independently of LVGL and has sufficient // compute for foreground announce verification/projection. Keeping the @@ -3266,6 +3347,7 @@ LxmfAdapter::RuntimeBudget LxmfAdapter::makeRuntimeBudget() const budget.allow_persistence = true; budget.allow_peer_projection = true; budget.allow_announce_tx = true; + budget.allow_propagation_client = true; budget.phase = "p4_screen"; return budget; #endif @@ -3276,7 +3358,9 @@ LxmfAdapter::RuntimeBudget LxmfAdapter::makeRuntimeBudget() const budget.allow_persistence = false; budget.allow_peer_projection = !screen_runtime::is_saver_active(); budget.allow_announce_tx = true; - budget.phase = screen_runtime::is_saver_active() ? "saver" : "screen"; + budget.allow_propagation_client = true; + budget.drop_public_discovery = true; + budget.phase = "screen"; return budget; } @@ -3329,7 +3413,10 @@ void LxmfAdapter::processRuntime() propagation_client_.cull(currentTimestampSeconds(), propagation_limits); pumpNomadPageRequests(); - processPropagationClient(); + if (budget.allow_propagation_client) + { + processPropagationClient(); + } processDeferredDiscoveryPackets(budget); if (budget.allow_peer_projection) @@ -3501,6 +3588,11 @@ bool LxmfAdapter::processOneRadioPacket( if (shouldDeferDiscoveryPacket(parsed, ingress_interface, budget)) { + if (deferred_replay) + { + noteRxSummary(false, false, false, false, true); + return false; + } if (!hasDeferredDiscoveryPacket(packet_hash) && enqueueDeferredDiscoveryPacket(rx_packet, packet_hash)) { @@ -3544,7 +3636,7 @@ bool LxmfAdapter::processOneRadioPacket( packet_len, parsed, ingress_interface, - budget.allow_persistence || deferred_replay); + budget.allow_persistence); case runtime::PacketRoute::Proof: return handleProofPacket(packet, packet_len, parsed, ingress_interface); case runtime::PacketRoute::LinkRequest: @@ -3916,7 +4008,7 @@ bool LxmfAdapter::lastAnnounceTxReachedRequiredInterfaces(bool sent) const { return false; } - return interfaces_.lastTxResult().reachedRequiredInterfaces(); + return interfaces_.lastTxResult().sent(); } bool LxmfAdapter::handleAnnouncePacket(const uint8_t* raw_packet, size_t raw_len, @@ -4551,7 +4643,11 @@ bool LxmfAdapter::handleLinkRequestPacket( if (local_kind == LocalDestinationKind::CallAudio && ingress_interface == reticulum::interfaces::InterfaceKind::LoRa) { - return false; + char call_hash[12] = {}; + formatHashPrefix(packet.destination_hash, call_hash, sizeof(call_hash)); + Serial.printf("[LXMF][CallRX] reject_bearer dest=%s reason=lora_not_supported\n", + call_hash); + return true; } if (!packet.payload || @@ -4702,27 +4798,9 @@ bool LxmfAdapter::handleLinkRequestPacket( return sendLinkHandshakeProof(*session); } - const PathEntry* path = - path_manager_.findPath(packet.destination_hash, millis(), kPathTtlMs); - if (!path) - { - return false; - } - - uint8_t link_id[reticulum::kTruncatedHashSize] = {}; - if (computeLinkIdFromLinkRequest(raw_packet, raw_len, packet, link_id)) - { - LinkRelayEntry& relay = - path_manager_.upsertLinkRelay(link_id, kMaxLinkRelays); - relay.initiator_interface_id = active_ingress_interface_id_; - relay.responder_interface_id = path->interface_id; - relay.initiator_hops = packet.hops; - relay.responder_hops = path->hops; - relay.last_seen_ms = millis(); - } - - return sendForwardPlan(packet, - packet_router_.planPathForward(*path, packet.hops)); + // Trail Mate is a Reticulum endpoint client. It should not become a + // transient transport router on LoRa and spend airtime forwarding links. + return false; } bool LxmfAdapter::handlePathRequestPacket(const reticulum::ParsedPacket& packet) @@ -4772,17 +4850,19 @@ bool LxmfAdapter::handlePathRequestPacket(const reticulum::ParsedPacket& packet) if (isLocalDestinationHash(requested_hash, &local_kind)) { (void)tag; + if (local_kind == LocalDestinationKind::CallAudio && + isLoRaInterfaceId(active_ingress_interface_id_)) + { + char call_hash[12] = {}; + formatHashPrefix(requested_hash, call_hash, sizeof(call_hash)); + Serial.printf("[LXMF][PathRX] skip_response dest=%s kind=call_audio reason=lora_not_supported\n", + call_hash); + return true; + } return sendAnnounce(local_kind, reticulum::PacketContext::PathResponse); } - const PathEntry* path = - path_manager_.findPath(requested_hash, millis(), kPathTtlMs); - if (!path || path->cached_announce_len == 0) - { - return true; - } - - return sendCachedAnnounceResponse(*path, reticulum::PacketContext::PathResponse); + return true; } bool LxmfAdapter::handleCacheRequestPacket(const reticulum::ParsedPacket& packet) @@ -6530,43 +6610,10 @@ bool LxmfAdapter::sendForwardPlan(const reticulum::ParsedPacket& packet, bool LxmfAdapter::maybeForwardTransportPacket(const uint8_t* raw_packet, size_t raw_len, const reticulum::ParsedPacket& packet) { - if (!raw_packet || raw_len == 0 || !packet.destination_hash) - { - return false; - } - - if (!packet.transport_id || - !hashesEqual(packet.transport_id, identity_.identityHash(), reticulum::kTruncatedHashSize)) - { - return false; - } - - if (isLocalDestinationHash(packet.destination_hash, nullptr)) - { - return false; - } - - const PathEntry* path = - path_manager_.findPath(packet.destination_hash, millis(), kPathTtlMs); - if (!path) - { - return false; - } - - if (packet.packet_type != reticulum::PacketType::Proof && - packet.packet_type != reticulum::PacketType::Announce) - { - uint8_t proof_hash[reticulum::kTruncatedHashSize] = {}; - reticulum::computeTruncatedPacketHash(raw_packet, raw_len, proof_hash); - path_manager_.rememberReversePath(proof_hash, - active_ingress_interface_id_, - path->hops, - millis(), - kMaxReverseEntries); - } - - return sendForwardPlan(packet, - packet_router_.planPathForward(*path, packet.hops)); + (void)raw_packet; + (void)raw_len; + (void)packet; + return false; } bool LxmfAdapter::maybeForwardLinkPacket(const uint8_t* raw_packet, size_t raw_len, @@ -6574,34 +6621,8 @@ bool LxmfAdapter::maybeForwardLinkPacket(const uint8_t* raw_packet, size_t raw_l { (void)raw_packet; (void)raw_len; - - if (!packet.destination_hash) - { - return false; - } - - if (packet.destination_type != reticulum::DestinationType::Link) - { - return false; - } - - LinkRelayEntry* relay = path_manager_.findLinkRelay(packet.destination_hash); - if (!relay) - { - return false; - } - - const runtime::PacketForwardPlan plan = - packet_router_.planLinkRelayForward(*relay, - active_ingress_interface_id_, - packet.hops); - if (!plan.forward) - { - return false; - } - - relay->last_seen_ms = millis(); - return sendForwardPlan(packet, plan); + (void)packet; + return false; } bool LxmfAdapter::sendProofForPacket(const uint8_t* raw_packet, size_t raw_len) @@ -6968,6 +6989,14 @@ bool LxmfAdapter::sendLinkRequest(LinkSession& session) bool routed = false; const PathEntry* tx_path = path_manager_.findPath(parsed.destination_hash, millis(), kPathTtlMs); + if (session.destination == LocalDestinationKind::CallAudio && + isLoRaPath(tx_path)) + { + Serial.printf("[LXMF][LinkTX] request_fail dest=%s kind=%u reason=call_lora_path\n", + dest_hash, + static_cast(session.destination)); + return false; + } if (tx_path && tx_path->hops > 1 && !tx_path->direct) { tx_packet_len = sizeof(link_request_routed_scratch_); @@ -7390,6 +7419,11 @@ bool LxmfAdapter::sendCachedAnnounceResponse(const PathEntry& path, { return false; } + if (isLoRaInterfaceId(active_ingress_interface_id_) && + isLxstTelephonyAnnouncePacket(parsed)) + { + return true; + } std::memset(routed_packet_scratch_, 0, sizeof(routed_packet_scratch_)); size_t packet_len = sizeof(routed_packet_scratch_); @@ -7706,27 +7740,9 @@ bool LxmfAdapter::shouldRebroadcastAnnounce( const reticulum::ParsedPacket& packet, reticulum::interfaces::InterfaceKind ingress_interface) const { - if (ingress_interface == reticulum::interfaces::InterfaceKind::WifiGateway) - { - return false; - } - if (!packet.destination_hash) - { - return false; - } - if (packet.context == static_cast(reticulum::PacketContext::PathResponse)) - { - return false; - } - if (packet.hops >= kMaxTransportHops) - { - return false; - } - if (hashesEqual(packet.destination_hash, identity_.destinationHash(), reticulum::kTruncatedHashSize)) - { - return false; - } - return true; + (void)packet; + (void)ingress_interface; + return false; } bool LxmfAdapter::rebroadcastAnnounce(const PathEntry& path, const reticulum::ParsedPacket& packet) diff --git a/platform/esp/arduino_common/src/chat/infra/meshtastic/mt_adapter.cpp b/platform/esp/arduino_common/src/chat/infra/meshtastic/mt_adapter.cpp index fd6dbfa5..8ae56a25 100644 --- a/platform/esp/arduino_common/src/chat/infra/meshtastic/mt_adapter.cpp +++ b/platform/esp/arduino_common/src/chat/infra/meshtastic/mt_adapter.cpp @@ -891,6 +891,15 @@ meshtastic_Routing_Error MtAdapter::getLastRoutingError() const void MtAdapter::setMqttProxySettings(const MqttProxySettings& settings) { const bool root_changed = mqtt_proxy_settings_.root != settings.root; + const bool downlink_policy_changed = + mqtt_proxy_settings_.enabled != settings.enabled || + mqtt_proxy_settings_.proxy_to_client_enabled != settings.proxy_to_client_enabled || + mqtt_proxy_settings_.encryption_enabled != settings.encryption_enabled || + mqtt_proxy_settings_.primary_downlink_enabled != settings.primary_downlink_enabled || + mqtt_proxy_settings_.secondary_downlink_enabled != settings.secondary_downlink_enabled || + mqtt_proxy_settings_.root != settings.root || + mqtt_proxy_settings_.primary_channel_id != settings.primary_channel_id || + mqtt_proxy_settings_.secondary_channel_id != settings.secondary_channel_id; const bool changed = mqtt_proxy_settings_.enabled != settings.enabled || mqtt_proxy_settings_.proxy_to_client_enabled != settings.proxy_to_client_enabled || mqtt_proxy_settings_.encryption_enabled != settings.encryption_enabled || @@ -910,6 +919,21 @@ void MtAdapter::setMqttProxySettings(const MqttProxySettings& settings) mqtt_proxy_settings_.root.c_str(), settings.root.c_str()); } + if (downlink_policy_changed) + { + const size_t dropped = mqtt_downlink_tx_queue_.size(); + mqtt_downlink_tx_queue_.clear(); + for (auto& entry : mqtt_downlink_seen_) + { + entry = MqttDownlinkSeenEntry{}; + } + mqtt_downlink_seen_next_ = 0; + if (dropped > 0) + { + LORA_LOG("[MQTT][DownlinkTX] settings changed drop pending=%u\n", + static_cast(dropped)); + } + } mqtt_proxy_settings_ = settings; if (changed) { @@ -1275,8 +1299,8 @@ bool MtAdapter::injectMqttEnvelope(const meshtastic_MeshPacket& packet, } else { - const bool tx_ok = transmitWirePacket(wire_buffer, wire_size); - LORA_LOG("[MQTT] downlink mesh tx id=%08lX ch=0x%02X len=%u ok=%u\n", + const bool tx_ok = enqueueMqttDownlinkTx(wire_buffer, wire_size, *tx_header); + LORA_LOG("[MQTT] downlink mesh tx scheduled id=%08lX ch=0x%02X len=%u ok=%u\n", (unsigned long)tx_header->id, (unsigned)tx_header->channel, (unsigned)wire_size, @@ -1290,6 +1314,145 @@ bool MtAdapter::injectMqttEnvelope(const meshtastic_MeshPacket& packet, return true; } +bool MtAdapter::isMqttDownlinkRecentlySeen(NodeId from, + MessageId msg_id, + uint8_t channel_hash, + uint32_t now_ms) +{ + for (auto& entry : mqtt_downlink_seen_) + { + if (!entry.used) + { + continue; + } + if (now_ms - entry.seen_ms > kMqttDownlinkSeenTtlMs) + { + entry = MqttDownlinkSeenEntry{}; + continue; + } + if (entry.from == from && + entry.msg_id == msg_id && + entry.channel_hash == channel_hash) + { + return true; + } + } + return false; +} + +void MtAdapter::rememberMqttDownlinkSeen(NodeId from, + MessageId msg_id, + uint8_t channel_hash, + uint32_t now_ms) +{ + for (auto& entry : mqtt_downlink_seen_) + { + if (entry.used && + entry.from == from && + entry.msg_id == msg_id && + entry.channel_hash == channel_hash) + { + entry.seen_ms = now_ms; + return; + } + } + + for (auto& entry : mqtt_downlink_seen_) + { + if (!entry.used || now_ms - entry.seen_ms > kMqttDownlinkSeenTtlMs) + { + entry.used = true; + entry.from = from; + entry.msg_id = msg_id; + entry.channel_hash = channel_hash; + entry.seen_ms = now_ms; + return; + } + } + + MqttDownlinkSeenEntry& entry = mqtt_downlink_seen_[mqtt_downlink_seen_next_]; + mqtt_downlink_seen_next_ = (mqtt_downlink_seen_next_ + 1U) % mqtt_downlink_seen_.size(); + entry.used = true; + entry.from = from; + entry.msg_id = msg_id; + entry.channel_hash = channel_hash; + entry.seen_ms = now_ms; +} + +bool MtAdapter::enqueueMqttDownlinkTx(const uint8_t* wire_data, + size_t wire_size, + const PacketHeaderWire& header) +{ + if (!wire_data || wire_size == 0 || wire_size > kMqttDownlinkWireMaxLen) + { + LORA_LOG("[MQTT][DownlinkTX] drop reason=invalid_wire id=%08lX len=%u\n", + static_cast(header.id), + static_cast(wire_size)); + return false; + } + + const uint32_t now_ms = millis(); + if (isMqttDownlinkRecentlySeen(header.from, header.id, header.channel, now_ms)) + { + LORA_LOG("[MQTT][DownlinkTX] skip reason=duplicate from=%08lX id=%08lX ch=0x%02X depth=%u\n", + static_cast(header.from), + static_cast(header.id), + static_cast(header.channel), + static_cast(mqtt_downlink_tx_queue_.size())); + return true; + } + + for (size_t i = 0; i < mqtt_downlink_tx_queue_.size(); ++i) + { + const PendingMqttDownlinkTx* pending = mqtt_downlink_tx_queue_.get(i); + if (pending && + pending->from == header.from && + pending->msg_id == header.id && + pending->channel_hash == header.channel) + { + rememberMqttDownlinkSeen(header.from, header.id, header.channel, now_ms); + LORA_LOG("[MQTT][DownlinkTX] skip reason=already_queued from=%08lX id=%08lX ch=0x%02X depth=%u\n", + static_cast(header.from), + static_cast(header.id), + static_cast(header.channel), + static_cast(mqtt_downlink_tx_queue_.size())); + return true; + } + } + + if (mqtt_downlink_tx_queue_.isFull()) + { + rememberMqttDownlinkSeen(header.from, header.id, header.channel, now_ms); + LORA_LOG("[MQTT][DownlinkTX] drop reason=pending_queue_full from=%08lX to=%08lX id=%08lX ch=0x%02X depth=%u\n", + static_cast(header.from), + static_cast(header.to), + static_cast(header.id), + static_cast(header.channel), + static_cast(mqtt_downlink_tx_queue_.size())); + return false; + } + + PendingMqttDownlinkTx pending{}; + std::memcpy(pending.wire.data(), wire_data, wire_size); + pending.wire_size = wire_size; + pending.from = header.from; + pending.to = header.to; + pending.msg_id = header.id; + pending.channel_hash = header.channel; + pending.first_seen_ms = now_ms; + + mqtt_downlink_tx_queue_.append(pending); + rememberMqttDownlinkSeen(header.from, header.id, header.channel, now_ms); + LORA_LOG("[MQTT][DownlinkTX] queued from=%08lX to=%08lX id=%08lX ch=0x%02X len=%u depth=%u\n", + static_cast(header.from), + static_cast(header.to), + static_cast(header.id), + static_cast(header.channel), + static_cast(wire_size), + static_cast(mqtt_downlink_tx_queue_.size())); + return true; +} + bool MtAdapter::handleMqttProxyMessage(const meshtastic_MqttClientProxyMessage& msg) { const bool is_data = msg.which_payload_variant == meshtastic_MqttClientProxyMessage_data_tag; @@ -1663,7 +1826,7 @@ void MtAdapter::processReceivedPacket(const uint8_t* data, size_t size) #endif // Check for duplicates - if (dedup_.isDuplicate(header.from, header.id)) + if (dedup_.isDuplicate(header.from, header.id, header.channel)) { mt_diag_dropf(&header, "dedup"); LORA_LOG("[LORA] RX dedup from=%08lX id=%08lX\n", @@ -1975,7 +2138,7 @@ void MtAdapter::processReceivedPacket(const uint8_t* data, size_t size) // Only mark packets as seen after we have successfully identified and decoded them. // This avoids poisoning dedup for retries when a packet failed due to stale PKI state. - dedup_.markSeen(header.from, header.id); + dedup_.markSeen(header.from, header.id, header.channel); if (plaintext_len > 0) { @@ -2631,6 +2794,96 @@ void MtAdapter::processSendQueue() } } } + + processMqttDownlinkTxQueue(now); +} + +void MtAdapter::processMqttDownlinkTxQueue(uint32_t now_ms) +{ + uint8_t drained = 0; + while (!mqtt_downlink_tx_queue_.empty() && + drained < kMqttDownlinkTxDrainPerTick) + { + PendingMqttDownlinkTx* pending = mqtt_downlink_tx_queue_.get(0); + if (!pending) + { + break; + } + + if (!config_.tx_enabled) + { + LORA_LOG("[MQTT][DownlinkTX] drop reason=tx_disabled from=%08lX id=%08lX ch=0x%02X depth=%u\n", + static_cast(pending->from), + static_cast(pending->msg_id), + static_cast(pending->channel_hash), + static_cast(mqtt_downlink_tx_queue_.size())); + PendingMqttDownlinkTx discarded{}; + mqtt_downlink_tx_queue_.popOldest(&discarded); + continue; + } + + if (min_tx_interval_ms_ > 0 && last_tx_ms_ > 0 && + (now_ms - last_tx_ms_) < min_tx_interval_ms_) + { + LORA_LOG("[MQTT][DownlinkTX] deferred reason=airtime_budget id=%08lX wait_ms=%lu depth=%u\n", + static_cast(pending->msg_id), + static_cast(min_tx_interval_ms_ - (now_ms - last_tx_ms_)), + static_cast(mqtt_downlink_tx_queue_.size())); + break; + } + + if (pending->retry_count > 0 && + (now_ms - pending->last_attempt_ms) < RETRY_DELAY_MS) + { + break; + } + + pending->last_attempt_ms = now_ms; + if (transmitWirePacket(pending->wire.data(), pending->wire_size)) + { + last_tx_ms_ = now_ms; + LORA_LOG("[MQTT][DownlinkTX] sent_to_radio from=%08lX to=%08lX id=%08lX ch=0x%02X len=%u retries=%u age_ms=%lu depth=%u\n", + static_cast(pending->from), + static_cast(pending->to), + static_cast(pending->msg_id), + static_cast(pending->channel_hash), + static_cast(pending->wire_size), + static_cast(pending->retry_count), + static_cast(now_ms - pending->first_seen_ms), + static_cast(mqtt_downlink_tx_queue_.size())); + PendingMqttDownlinkTx discarded{}; + mqtt_downlink_tx_queue_.popOldest(&discarded); + ++drained; + continue; + } + + ++pending->retry_count; + const char* reason = board_.isRadioOnline() ? "radio_queue_full" : "radio_offline"; + if (pending->retry_count > kMqttDownlinkTxMaxRetries) + { + LORA_LOG("[MQTT][DownlinkTX] drop reason=%s from=%08lX to=%08lX id=%08lX ch=0x%02X retries=%u age_ms=%lu depth=%u\n", + reason, + static_cast(pending->from), + static_cast(pending->to), + static_cast(pending->msg_id), + static_cast(pending->channel_hash), + static_cast(pending->retry_count), + static_cast(now_ms - pending->first_seen_ms), + static_cast(mqtt_downlink_tx_queue_.size())); + PendingMqttDownlinkTx discarded{}; + mqtt_downlink_tx_queue_.popOldest(&discarded); + continue; + } + + LORA_LOG("[MQTT][DownlinkTX] deferred reason=%s from=%08lX id=%08lX ch=0x%02X retries=%u depth=%u\n", + reason, + static_cast(pending->from), + static_cast(pending->msg_id), + static_cast(pending->channel_hash), + static_cast(pending->retry_count), + static_cast(mqtt_downlink_tx_queue_.size())); + break; + } } bool MtAdapter::sendPacket(const PendingSend& pending) diff --git a/platform/esp/common/src/reticulum_call_audio_engine.cpp b/platform/esp/common/src/reticulum_call_audio_engine.cpp index cbef3107..3b6ba627 100644 --- a/platform/esp/common/src/reticulum_call_audio_engine.cpp +++ b/platform/esp/common/src/reticulum_call_audio_engine.cpp @@ -29,6 +29,7 @@ constexpr uint8_t kJitterStartFrames = 3; constexpr uint8_t kJitterMaxFrames = 32; constexpr uint8_t kInboundDrainLimit = 4; constexpr uint8_t kIoFailureLimit = 25; +constexpr uint8_t kCallSpeakerVolumePercent = 100; constexpr float kTxPcmGain = 1.0f; constexpr float kRxPcmGain = 1.0f; constexpr uint32_t kCaptureTaskStackBytes = 20 * 1024; @@ -590,6 +591,7 @@ bool media_start() return false; } s_backend_open = true; + set_speaker_volume(kCallSpeakerVolumePercent); const auto snapshot = ::platform::ui::reticulum_call::snapshot(); const CodecSelection selection = codec_selection(snapshot); diff --git a/platform/esp/radio/meshtastic_radio_adapter.cpp b/platform/esp/radio/meshtastic_radio_adapter.cpp index 84f7396c..c1a038be 100644 --- a/platform/esp/radio/meshtastic_radio_adapter.cpp +++ b/platform/esp/radio/meshtastic_radio_adapter.cpp @@ -442,7 +442,7 @@ void MeshtasticRadioAdapter::processReceivedPacket(const uint8_t* data, size_t s static_cast(header.id)); return; } - if (dedup_.isDuplicate(header.from, header.id)) + if (dedup_.isDuplicate(header.from, header.id, header.channel)) { ESP_LOGI(kTag, "rx drop duplicate from=%08lX id=%08lX", @@ -450,7 +450,7 @@ void MeshtasticRadioAdapter::processReceivedPacket(const uint8_t* data, size_t s static_cast(header.id)); return; } - dedup_.markSeen(header.from, header.id); + dedup_.markSeen(header.from, header.id, header.channel); const chat::ChannelId channel = channel_from_hash(header.channel, primary_channel_hash_, secondary_channel_hash_);