Implement Reticulum runtime budget policy

This commit is contained in:
liu weikai committed 2026-07-09 00:25:10 +08:00
1 parent f825c7e276
commit 12ec1f77cf
7 files changed
+572 -160

No files matched your search

+3
View File
@@ -24,6 +24,9 @@ jobs:
- name: Verify Wi-Fi access policy
run: python3 scripts/check_wifi_access_policy.py
- name: Verify Reticulum runtime budget policy
run: python3 scripts/check_reticulum_runtime_budget_policy.py
- name: Verify track file streaming
run: python3 scripts/check_track_file_streaming.py
+66
View File
@@ -0,0 +1,66 @@
# Protocol Runtime Budget Policy
Trail Mate treats mesh protocols as device-terminal runtimes, not desktop
routers. Protocol adapters must protect the UI, LoRa radio, serial logging, SD
card, and NVS budgets before trying to ingest background network traffic.
## Reticulum Runtime Model
Reticulum runs as a low-frequency terminal ingress runtime.
When the screen is on, realtime processing is limited to:
- LXMF direct traffic addressed to this device.
- LXMF traffic addressed to configured Reticulum groups.
- Path, proof, link, and cache traffic that this device explicitly requested or
that is required by an active local session.
- Outbound user actions such as sending text, app data, and self announces.
Ordinary public announces and public discovery traffic are not realtime while
the user is interacting with the device. They may be delayed for seconds or
tens of seconds and replayed during an idle or screen-off maintenance window.
## Required Boundaries
- `pollIncomingText()` and `pollIncomingData()` may only return already
materialised Trail Mate business messages. They must not parse Reticulum
packets, verify announces, write SD files, persist peer caches, or send
announces.
- Reticulum packet ingestion must run from a periodic runtime pump with an
explicit budget. On ESP Arduino this is currently carried by the adapter
`processSendQueue()` hook because it already runs outside the LVGL loop.
- LoRa and Wi-Fi Reticulum ingress must share the same public-discovery budget
rules. Wi-Fi-only discovery throttles are not sufficient.
- Public discovery persistence must be deferred. `record_announce()`,
`record_lxmf_address()`, and peer cache persistence must not run from the
screen-on RX hot path for ordinary public announces.
- Peer cache persistence is dirty/coalesced. Code must not force
`maybePersistPeers(true)` from RX paths. Maintenance windows decide when the
dirty peer cache is written.
- RX hot-path logging must be summary-first. Detailed logs are acceptable for
local/realtime traffic and diagnostics, but public discovery should normally
be represented by periodic counters.
## Maintenance Windows
A maintenance window is available when the screen runtime reports that the
device is sleeping and the saver is not active. During maintenance windows the
Reticulum runtime may:
- Replay deferred discovery packets under the discovery sample budget.
- Persist Reticulum announces and LXMF address records to SD.
- Flush dirty peer cache state to NVS.
- Publish deferred peer projections to the contact store.
The maintenance window still has finite budgets. It must not drain unbounded
network queues in one cycle.
## Regression Checks
`scripts/check_reticulum_runtime_budget_policy.py` enforces the highest-risk
boundaries:
- Reticulum polling APIs must not call the packet ingestion pump.
- Reticulum RX paths must not force peer persistence.
- Discovery budget names and helpers must not regress to Wi-Fi-only concepts.
- The Reticulum product adapter must forward the periodic runtime pump.
@@ -12,6 +12,7 @@
#include "platform/esp/arduino_common/chat/infra/lxmf/lxmf_identity.h"
#include "platform/esp/arduino_common/chat/infra/lxmf/lxmf_runtime_state.h"
#include "platform/esp/arduino_common/chat/infra/reticulum/reticulum_interfaces.h"
#include "sys/ringbuf.h"
#include <array>
#include <cstddef>
@@ -57,6 +58,7 @@ class LxmfAdapter : public IMeshAdapter
bool pollIncomingRawPacket(uint8_t* out_data, size_t& out_len, size_t max_len) override;
void handleRawPacket(const uint8_t* data, size_t size) override;
void setLastRxStats(float rssi, float snr) override;
void processSendQueue() override;
private:
using PeerInfo = runtime::PeerInfo;
@@ -79,15 +81,39 @@ class LxmfAdapter : public IMeshAdapter
static constexpr uint32_t kAnnounceIntervalMs = 120000;
static constexpr uint32_t kInitialAnnounceDelayMs = 1500;
static constexpr uint8_t kMaxIngressPacketsPerPoll = 2;
static constexpr uint32_t kWifiDiscoverySampleIntervalMs = 10000;
static constexpr uint32_t kDiscoverySampleIntervalMs = 10000;
static constexpr uint32_t kRxSummaryIntervalMs = 5000;
static constexpr uint32_t kAnnounceRebroadcastIntervalMs = 60000;
static constexpr uint32_t kPeerProjectionSleepIntervalMs = 250;
static constexpr std::size_t kPendingPeerProjectionDepth = 24;
static constexpr uint32_t kPeerPersistSleepIntervalMs = 15000;
static constexpr std::size_t kDeferredDiscoveryDepth = 8;
struct RuntimeBudget
{
uint8_t live_packet_limit = 1;
uint8_t deferred_discovery_limit = 0;
bool allow_public_discovery = false;
bool allow_persistence = false;
bool allow_peer_projection = false;
bool allow_announce_tx = true;
const char* phase = "screen";
};
struct DeferredDiscoveryPacket
{
uint8_t data[reticulum::kReticulumMtu] = {};
size_t len = 0;
RxMeta rx_meta{};
reticulum::interfaces::InterfaceKind interface_kind =
reticulum::interfaces::InterfaceKind::LoRa;
uint8_t packet_hash[reticulum::kFullHashSize] = {};
};
reticulum::interfaces::ReticulumInterfaceSet interfaces_;
reticulum::interfaces::RxPacket rx_packet_scratch_{};
sys::RingBuffer<DeferredDiscoveryPacket, kDeferredDiscoveryDepth> deferred_discovery_queue_;
DeferredDiscoveryPacket deferred_discovery_scratch_{};
LxmfIdentity identity_;
MeshConfig config_{};
static constexpr std::size_t kIncomingQueueDepth = 4;
@@ -100,6 +126,7 @@ class LxmfAdapter : public IMeshAdapter
std::string user_long_name_;
std::string user_short_name_;
uint32_t last_announce_ms_ = 0;
uint32_t last_lora_discovery_sample_ms_ = 0;
uint32_t last_wifi_discovery_sample_ms_ = 0;
uint32_t last_rx_summary_ms_ = 0;
uint32_t last_announce_rebroadcast_ms_ = 0;
@@ -107,6 +134,8 @@ class LxmfAdapter : public IMeshAdapter
uint32_t rx_summary_wifi_skipped_ = 0;
uint32_t rx_summary_duplicates_ = 0;
uint32_t rx_summary_parse_failed_ = 0;
uint32_t rx_summary_deferred_ = 0;
uint32_t rx_summary_deferred_dropped_ = 0;
std::array<NodeId, kPendingPeerProjectionDepth> pending_peer_projection_nodes_{};
std::size_t pending_peer_projection_count_ = 0;
uint32_t last_peer_projection_ms_ = 0;
@@ -115,14 +144,33 @@ class LxmfAdapter : public IMeshAdapter
bool announce_pending_ = true;
bool peers_loaded_ = false;
bool peer_persist_dirty_ = false;
RxMeta active_rx_meta_{};
bool has_active_rx_meta_ = false;
void processRadioPackets();
RuntimeBudget makeRuntimeBudget() const;
void processRuntime();
void processRadioPackets(const RuntimeBudget& budget);
bool processOneRadioPacket(const reticulum::interfaces::RxPacket& packet,
const RuntimeBudget& budget,
bool deferred_replay);
bool shouldDeferDiscoveryPacket(
const reticulum::ParsedPacket& packet,
reticulum::interfaces::InterfaceKind ingress_interface,
const RuntimeBudget& budget);
bool isPublicDiscoveryPacket(const reticulum::ParsedPacket& packet) const;
bool enqueueDeferredDiscoveryPacket(
const reticulum::interfaces::RxPacket& packet,
const uint8_t packet_hash[reticulum::kFullHashSize]);
bool hasDeferredDiscoveryPacket(
const uint8_t packet_hash[reticulum::kFullHashSize]) const;
void processDeferredDiscoveryPackets(const RuntimeBudget& budget);
void maybeAnnounce();
bool sendAnnounce(LocalDestinationKind kind = LocalDestinationKind::Delivery,
reticulum::PacketContext context = reticulum::PacketContext::None);
bool handleAnnouncePacket(const uint8_t* raw_packet, size_t raw_len,
const reticulum::ParsedPacket& packet,
reticulum::interfaces::InterfaceKind ingress_interface);
reticulum::interfaces::InterfaceKind ingress_interface,
bool allow_persistence);
bool handleDataPacket(const uint8_t* raw_packet, size_t raw_len,
const reticulum::ParsedPacket& packet);
bool handleProofPacket(const uint8_t* raw_packet, size_t raw_len,
@@ -164,10 +212,12 @@ class LxmfAdapter : public IMeshAdapter
bool shouldProcessWifiIngressPacket(const reticulum::ParsedPacket& packet);
bool shouldLogRxDetail(const reticulum::ParsedPacket& packet,
reticulum::interfaces::InterfaceKind ingress_interface) const;
bool consumeWifiDiscoveryBudget();
bool consumeDiscoveryBudget(reticulum::interfaces::InterfaceKind ingress_interface);
void noteRxSummary(bool wifi_skipped = false,
bool duplicate = false,
bool parse_failed = false);
bool parse_failed = false,
bool deferred = false,
bool deferred_dropped = false);
bool shouldRebroadcastAnnounce(
const reticulum::ParsedPacket& packet,
reticulum::interfaces::InterfaceKind ingress_interface) const;
@@ -63,6 +63,7 @@ class ReticulumAdapter final : public IMeshAdapter
bool pollIncomingRawPacket(uint8_t* out_data, size_t& out_len, size_t max_len) override;
void handleRawPacket(const uint8_t* data, size_t size) override;
void setLastRxStats(float rssi, float snr) override;
void processSendQueue() override;
private:
std::unique_ptr<lxmf::LxmfAdapter> service_;
@@ -691,8 +691,6 @@ MeshSendResult LxmfAdapter::sendTextDetailed(ChannelId channel,
{
(void)forced_msg_id;
processRadioPackets();
Serial.printf("[LXMF][DirectTX] begin ch=%u peer=%08lX len=%u ready=%u\n",
static_cast<unsigned>(channel),
static_cast<unsigned long>(peer),
@@ -855,8 +853,6 @@ MeshSendResult LxmfAdapter::sendTextToReticulumDestination(
MessageId forced_msg_id,
const ReticulumPeerIdentity& destination)
{
processRadioPackets();
char dest_hash[12] = {};
formatHashPrefix(destination.destination_hash, dest_hash, sizeof(dest_hash));
Serial.printf("[LXMF][GroupTX] begin ch=%u forced=%lu dest=%s len=%u ready=%u\n",
@@ -946,9 +942,6 @@ MeshSendResult LxmfAdapter::sendTextToReticulumDestination(
bool LxmfAdapter::pollIncomingText(MeshIncomingText* out)
{
processRadioPackets();
maybeAnnounce();
return text_receive_queue_.pop(out);
}
@@ -960,8 +953,6 @@ bool LxmfAdapter::sendAppData(ChannelId channel, uint32_t portnum,
{
(void)want_ack;
processRadioPackets();
Serial.printf("[LXMF][AppDataTX] begin ch=%u port=%lu dest=%08lX len=%u ready=%u want_ack=%u want_response=%u\n",
static_cast<unsigned>(channel),
static_cast<unsigned long>(portnum),
@@ -1171,9 +1162,6 @@ bool LxmfAdapter::sendAppData(ChannelId channel, uint32_t portnum,
bool LxmfAdapter::pollIncomingData(MeshIncomingData* out)
{
processRadioPackets();
maybeAnnounce();
return data_receive_queue_.pop(out);
}
@@ -1183,7 +1171,6 @@ bool LxmfAdapter::requestNodeInfo(NodeId dest, bool want_response)
if (dest != 0 && dest != 0xFFFFFFFFUL)
{
processRadioPackets();
PeerInfo* peer = findPeerByNodeId(dest);
if (!peer)
{
@@ -1325,9 +1312,41 @@ void LxmfAdapter::setLastRxStats(float rssi, float snr)
interfaces_.setLastRxStats(rssi, snr);
}
void LxmfAdapter::processRadioPackets()
void LxmfAdapter::processSendQueue()
{
pumpPendingPeerUpdates();
processRuntime();
}
LxmfAdapter::RuntimeBudget LxmfAdapter::makeRuntimeBudget() const
{
RuntimeBudget budget{};
const bool maintenance_window =
screen_runtime::is_sleeping() && !screen_runtime::is_saver_active();
if (maintenance_window)
{
budget.live_packet_limit = kMaxIngressPacketsPerPoll;
budget.deferred_discovery_limit = 2;
budget.allow_public_discovery = true;
budget.allow_persistence = true;
budget.allow_peer_projection = true;
budget.allow_announce_tx = true;
budget.phase = "sleep";
return budget;
}
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 = true;
budget.phase = screen_runtime::is_saver_active() ? "saver" : "screen";
return budget;
}
void LxmfAdapter::processRuntime()
{
const RuntimeBudget budget = makeRuntimeBudget();
cullTransportState();
cullLinkSessions();
const runtime::PropagationRuntimeLimits propagation_limits{
@@ -1341,114 +1360,256 @@ void LxmfAdapter::processRadioPackets()
currentTimestampSeconds(),
propagation_limits);
processRadioPackets(budget);
processDeferredDiscoveryPackets(budget);
if (budget.allow_peer_projection)
{
pumpPendingPeerUpdates();
}
if (budget.allow_persistence && peer_persist_dirty_)
{
(void)maybePersistPeers(false);
}
if (budget.allow_announce_tx)
{
maybeAnnounce();
}
}
void LxmfAdapter::processRadioPackets(const RuntimeBudget& budget)
{
uint8_t polled_packets = 0;
while (polled_packets < kMaxIngressPacketsPerPoll &&
while (polled_packets < budget.live_packet_limit &&
interfaces_.pollIncomingPacket(&rx_packet_scratch_))
{
++polled_packets;
const uint8_t* packet = rx_packet_scratch_.data;
const size_t packet_len = rx_packet_scratch_.len;
const auto ingress_interface = rx_packet_scratch_.interface_kind;
const bool ingress_wifi =
ingress_interface == reticulum::interfaces::InterfaceKind::WifiGateway;
const char* iface_label =
ingress_wifi ? "wifi" : "lora";
reticulum::ParsedPacket parsed{};
if (reticulum::parsePacket(packet, packet_len, &parsed))
(void)processOneRadioPacket(rx_packet_scratch_, budget, false);
}
MeshIncomingData discarded;
while (interfaces_.pollLegacyIncomingData(&discarded))
{
}
}
bool LxmfAdapter::processOneRadioPacket(
const reticulum::interfaces::RxPacket& rx_packet,
const RuntimeBudget& budget,
bool deferred_replay)
{
const uint8_t* packet = rx_packet.data;
const size_t packet_len = rx_packet.len;
const auto ingress_interface = rx_packet.interface_kind;
const bool ingress_wifi =
ingress_interface == reticulum::interfaces::InterfaceKind::WifiGateway;
const char* iface_label = ingress_wifi ? "wifi" : "lora";
active_rx_meta_ = rx_packet.rx_meta;
has_active_rx_meta_ = true;
struct ActiveRxMetaScope
{
bool& active;
~ActiveRxMetaScope() { active = false; }
} active_rx_meta_scope{has_active_rx_meta_};
reticulum::ParsedPacket parsed{};
if (!reticulum::parsePacket(packet, packet_len, &parsed))
{
noteRxSummary(false, false, true);
if (!ingress_wifi)
{
char dest_hash[12] = {};
formatHashPrefix(parsed.destination_hash, dest_hash, sizeof(dest_hash));
noteRxSummary();
const bool log_detail = shouldLogRxDetail(parsed, ingress_interface);
if (log_detail)
{
Serial.printf("[LXMF][RawRX] packet iface=%s type=%u dest_type=%u context=%u dest=%s raw_len=%u payload_len=%u hops=%u\n",
iface_label,
static_cast<unsigned>(parsed.packet_type),
static_cast<unsigned>(parsed.destination_type),
static_cast<unsigned>(parsed.context),
dest_hash,
static_cast<unsigned>(packet_len),
static_cast<unsigned>(parsed.payload_len),
static_cast<unsigned>(parsed.hops));
}
if (parsed.hops < 0xFF)
{
parsed.hops += 1;
}
uint8_t packet_hash[reticulum::kFullHashSize] = {};
reticulum::computePacketHash(packet, packet_len, packet_hash);
if (isDuplicatePacket(packet_hash))
{
noteRxSummary(false, true, false);
if (!ingress_wifi)
{
Serial.printf("[LXMF][RawRX] drop reason=duplicate dest=%s\n", dest_hash);
}
continue;
}
rememberPacket(packet_hash);
if (ingress_wifi && !shouldProcessWifiIngressPacket(parsed))
{
noteRxSummary(true, false, false);
continue;
}
if (parsed.packet_type == reticulum::PacketType::Announce)
{
handleAnnouncePacket(packet,
packet_len,
parsed,
ingress_interface);
}
else if (parsed.packet_type == reticulum::PacketType::Proof)
{
handleProofPacket(packet, packet_len, parsed);
}
else if (parsed.packet_type == reticulum::PacketType::LinkRequest)
{
handleLinkRequestPacket(packet, packet_len, parsed);
}
else if (parsed.packet_type == reticulum::PacketType::Data)
{
if (!handlePathRequestPacket(parsed) &&
!handleCacheRequestPacket(parsed) &&
!handleLocalLinkPacket(packet, packet_len, parsed) &&
!maybeForwardLinkPacket(packet, packet_len, parsed) &&
!maybeForwardTransportPacket(packet, packet_len, parsed))
{
handleDataPacket(packet, packet_len, parsed);
}
}
else
{
(void)handleLocalLinkPacket(packet, packet_len, parsed);
(void)maybeForwardLinkPacket(packet, packet_len, parsed);
(void)maybeForwardTransportPacket(packet, packet_len, parsed);
}
}
else
{
noteRxSummary(false, false, true);
if (rx_packet_scratch_.interface_kind != reticulum::interfaces::InterfaceKind::WifiGateway)
{
Serial.printf("[LXMF][RawRX] drop reason=parse_failed raw_len=%u first=%02X\n",
static_cast<unsigned>(packet_len),
static_cast<unsigned>(packet_len > 0 ? packet[0] : 0));
}
Serial.printf("[LXMF][RawRX] drop reason=parse_failed raw_len=%u first=%02X\n",
static_cast<unsigned>(packet_len),
static_cast<unsigned>(packet_len > 0 ? packet[0] : 0));
}
return false;
}
MeshIncomingData discarded;
while (interfaces_.pollLegacyIncomingData(&discarded))
char dest_hash[12] = {};
formatHashPrefix(parsed.destination_hash, dest_hash, sizeof(dest_hash));
noteRxSummary();
const bool log_detail = shouldLogRxDetail(parsed, ingress_interface) && !deferred_replay;
if (log_detail)
{
Serial.printf("[LXMF][RawRX] packet iface=%s type=%u dest_type=%u context=%u dest=%s raw_len=%u payload_len=%u hops=%u phase=%s\n",
iface_label,
static_cast<unsigned>(parsed.packet_type),
static_cast<unsigned>(parsed.destination_type),
static_cast<unsigned>(parsed.context),
dest_hash,
static_cast<unsigned>(packet_len),
static_cast<unsigned>(parsed.payload_len),
static_cast<unsigned>(parsed.hops),
budget.phase);
}
if (parsed.hops < 0xFF)
{
parsed.hops += 1;
}
uint8_t packet_hash[reticulum::kFullHashSize] = {};
reticulum::computePacketHash(packet, packet_len, packet_hash);
if (isDuplicatePacket(packet_hash))
{
noteRxSummary(false, true, false);
if (!ingress_wifi && !deferred_replay)
{
Serial.printf("[LXMF][RawRX] drop reason=duplicate dest=%s\n", dest_hash);
}
return false;
}
if (shouldDeferDiscoveryPacket(parsed, ingress_interface, budget))
{
if (!hasDeferredDiscoveryPacket(packet_hash) &&
enqueueDeferredDiscoveryPacket(rx_packet, packet_hash))
{
noteRxSummary(false, false, false, true, false);
return true;
}
noteRxSummary(false, false, false, true, true);
return false;
}
rememberPacket(packet_hash);
if (!deferred_replay && ingress_wifi && !shouldProcessWifiIngressPacket(parsed))
{
noteRxSummary(true, false, false);
return false;
}
if (parsed.packet_type == reticulum::PacketType::Announce)
{
return handleAnnouncePacket(packet,
packet_len,
parsed,
ingress_interface,
budget.allow_persistence || deferred_replay);
}
if (parsed.packet_type == reticulum::PacketType::Proof)
{
return handleProofPacket(packet, packet_len, parsed);
}
if (parsed.packet_type == reticulum::PacketType::LinkRequest)
{
return handleLinkRequestPacket(packet, packet_len, parsed);
}
if (parsed.packet_type == reticulum::PacketType::Data)
{
if (!handlePathRequestPacket(parsed) &&
!handleCacheRequestPacket(parsed) &&
!handleLocalLinkPacket(packet, packet_len, parsed) &&
!maybeForwardLinkPacket(packet, packet_len, parsed) &&
!maybeForwardTransportPacket(packet, packet_len, parsed))
{
return handleDataPacket(packet, packet_len, parsed);
}
return true;
}
const bool handled_local = handleLocalLinkPacket(packet, packet_len, parsed);
const bool handled_link = maybeForwardLinkPacket(packet, packet_len, parsed);
const bool handled_transport = maybeForwardTransportPacket(packet, packet_len, parsed);
return handled_local || handled_link || handled_transport;
}
bool LxmfAdapter::shouldDeferDiscoveryPacket(
const reticulum::ParsedPacket& packet,
reticulum::interfaces::InterfaceKind ingress_interface,
const RuntimeBudget& budget)
{
if (!isPublicDiscoveryPacket(packet))
{
return false;
}
if (!budget.allow_public_discovery)
{
return true;
}
return !consumeDiscoveryBudget(ingress_interface);
}
bool LxmfAdapter::isPublicDiscoveryPacket(const reticulum::ParsedPacket& packet) const
{
if (packet.packet_type != reticulum::PacketType::Announce ||
!packet.destination_hash)
{
return false;
}
if (isLocalDestinationHash(packet.destination_hash, nullptr))
{
return false;
}
return packet.context != static_cast<uint8_t>(reticulum::PacketContext::PathResponse) ||
!findPendingPathRequest(packet.destination_hash);
}
bool LxmfAdapter::enqueueDeferredDiscoveryPacket(
const reticulum::interfaces::RxPacket& packet,
const uint8_t packet_hash[reticulum::kFullHashSize])
{
if (!packet_hash || packet.len == 0 || packet.len > reticulum::kReticulumMtu)
{
return false;
}
deferred_discovery_scratch_ = DeferredDiscoveryPacket{};
DeferredDiscoveryPacket& deferred = deferred_discovery_scratch_;
memcpy(deferred.data, packet.data, packet.len);
deferred.len = packet.len;
deferred.rx_meta = packet.rx_meta;
deferred.interface_kind = packet.interface_kind;
memcpy(deferred.packet_hash, packet_hash, sizeof(deferred.packet_hash));
bool dropped = false;
deferred_discovery_queue_.pushDropOldest(deferred, &dropped);
if (dropped)
{
noteRxSummary(false, false, false, false, true);
}
return true;
}
bool LxmfAdapter::hasDeferredDiscoveryPacket(
const uint8_t packet_hash[reticulum::kFullHashSize]) const
{
if (!packet_hash)
{
return false;
}
for (std::size_t i = 0; i < deferred_discovery_queue_.size(); ++i)
{
const DeferredDiscoveryPacket* queued = deferred_discovery_queue_.get(i);
if (queued &&
hashesEqual(queued->packet_hash, packet_hash, reticulum::kFullHashSize))
{
return true;
}
}
pumpPendingPeerUpdates();
if (peer_persist_dirty_)
return false;
}
void LxmfAdapter::processDeferredDiscoveryPackets(const RuntimeBudget& budget)
{
if (!budget.allow_public_discovery || budget.deferred_discovery_limit == 0)
{
(void)maybePersistPeers(false);
return;
}
uint8_t processed = 0;
while (processed < budget.deferred_discovery_limit &&
deferred_discovery_queue_.popOldest(&deferred_discovery_scratch_))
{
rx_packet_scratch_.len = deferred_discovery_scratch_.len;
memcpy(rx_packet_scratch_.data,
deferred_discovery_scratch_.data,
deferred_discovery_scratch_.len);
rx_packet_scratch_.rx_meta = deferred_discovery_scratch_.rx_meta;
rx_packet_scratch_.interface_kind = deferred_discovery_scratch_.interface_kind;
(void)processOneRadioPacket(rx_packet_scratch_, budget, true);
++processed;
}
}
@@ -1611,7 +1772,8 @@ bool LxmfAdapter::sendAnnounce(LocalDestinationKind kind,
bool LxmfAdapter::handleAnnouncePacket(const uint8_t* raw_packet, size_t raw_len,
const reticulum::ParsedPacket& packet,
reticulum::interfaces::InterfaceKind ingress_interface)
reticulum::interfaces::InterfaceKind ingress_interface,
bool allow_persistence)
{
if (!raw_packet || raw_len == 0 || !packet.destination_hash)
{
@@ -1787,12 +1949,15 @@ bool LxmfAdapter::handleAnnouncePacket(const uint8_t* raw_packet, size_t raw_len
directory_announce.raw_packet_len = raw_len;
directory_announce.app_data = announce.app_data;
directory_announce.app_data_len = announce.app_data_len;
const auto announce_store_status = rtdir::record_announce(directory_announce);
if (announce_store_status.sd_present && !announce_store_status.saved)
if (allow_persistence)
{
Serial.printf("[LXMF][AnnounceRX] directory_save failed message=%s detail=%s\n",
announce_store_status.message,
announce_store_status.detail);
const auto announce_store_status = rtdir::record_announce(directory_announce);
if (announce_store_status.sd_present && !announce_store_status.saved)
{
Serial.printf("[LXMF][AnnounceRX] directory_save failed message=%s detail=%s\n",
announce_store_status.message,
announce_store_status.detail);
}
}
const bool log_announce_detail =
@@ -1870,7 +2035,7 @@ bool LxmfAdapter::handleAnnouncePacket(const uint8_t* raw_packet, size_t raw_len
identity_changed ||
display_changed ||
address_refresh_due;
if (should_store_address)
if (allow_persistence && should_store_address)
{
rtdir::LxmfAddressRecord directory_address{};
directory_address.valid = true;
@@ -1904,8 +2069,15 @@ bool LxmfAdapter::handleAnnouncePacket(const uint8_t* raw_packet, size_t raw_len
}
else
{
publishPeerUpdate(peer);
(void)maybePersistPeers(true);
if (allow_persistence)
{
publishPeerUpdate(peer);
}
else
{
queuePeerUpdate(peer);
}
(void)maybePersistPeers(false);
}
Serial.printf("[LXMF][AnnounceRX] learned peer=%08lX dest=%s identity=%s name=%s peers=%u\n",
static_cast<unsigned long>(peer.node_id),
@@ -4010,8 +4182,7 @@ bool LxmfAdapter::shouldProcessWifiIngressPacket(const reticulum::ParsedPacket&
return true;
}
return screen_runtime::is_sleeping() &&
!screen_runtime::is_saver_active() &&
consumeWifiDiscoveryBudget();
!screen_runtime::is_saver_active();
}
return false;
@@ -4041,23 +4212,30 @@ bool LxmfAdapter::shouldLogRxDetail(
return packet.destination_type == reticulum::DestinationType::Link;
}
bool LxmfAdapter::consumeWifiDiscoveryBudget()
bool LxmfAdapter::consumeDiscoveryBudget(
reticulum::interfaces::InterfaceKind ingress_interface)
{
uint32_t& last_sample_ms =
ingress_interface == reticulum::interfaces::InterfaceKind::WifiGateway
? last_wifi_discovery_sample_ms_
: last_lora_discovery_sample_ms_;
const uint32_t now_ms = millis();
if (last_wifi_discovery_sample_ms_ != 0 &&
(now_ms - last_wifi_discovery_sample_ms_) < kWifiDiscoverySampleIntervalMs)
if (last_sample_ms != 0 &&
(now_ms - last_sample_ms) < kDiscoverySampleIntervalMs)
{
return false;
}
last_wifi_discovery_sample_ms_ = now_ms;
last_sample_ms = now_ms;
return true;
}
void LxmfAdapter::noteRxSummary(bool wifi_skipped,
bool duplicate,
bool parse_failed)
bool parse_failed,
bool deferred,
bool deferred_dropped)
{
if (!wifi_skipped && !duplicate && !parse_failed)
if (!wifi_skipped && !duplicate && !parse_failed && !deferred && !deferred_dropped)
{
++rx_summary_packets_;
}
@@ -4073,6 +4251,14 @@ void LxmfAdapter::noteRxSummary(bool wifi_skipped,
{
++rx_summary_parse_failed_;
}
if (deferred)
{
++rx_summary_deferred_;
}
if (deferred_dropped)
{
++rx_summary_deferred_dropped_;
}
const uint32_t now_ms = millis();
if (last_rx_summary_ms_ == 0)
@@ -4087,21 +4273,27 @@ void LxmfAdapter::noteRxSummary(bool wifi_skipped,
if (rx_summary_packets_ == 0 &&
rx_summary_wifi_skipped_ == 0 &&
rx_summary_duplicates_ == 0 &&
rx_summary_parse_failed_ == 0)
rx_summary_parse_failed_ == 0 &&
rx_summary_deferred_ == 0 &&
rx_summary_deferred_dropped_ == 0)
{
last_rx_summary_ms_ = now_ms;
return;
}
Serial.printf("[LXMF][RawRX] stats packets=%u wifi_skipped=%u duplicate=%u parse_failed=%u\n",
Serial.printf("[LXMF][RawRX] stats packets=%u wifi_skipped=%u duplicate=%u parse_failed=%u deferred=%u deferred_drop=%u\n",
static_cast<unsigned>(rx_summary_packets_),
static_cast<unsigned>(rx_summary_wifi_skipped_),
static_cast<unsigned>(rx_summary_duplicates_),
static_cast<unsigned>(rx_summary_parse_failed_));
static_cast<unsigned>(rx_summary_parse_failed_),
static_cast<unsigned>(rx_summary_deferred_),
static_cast<unsigned>(rx_summary_deferred_dropped_));
rx_summary_packets_ = 0;
rx_summary_wifi_skipped_ = 0;
rx_summary_duplicates_ = 0;
rx_summary_parse_failed_ = 0;
rx_summary_deferred_ = 0;
rx_summary_deferred_dropped_ = 0;
last_rx_summary_ms_ = now_ms;
}
@@ -5002,31 +5194,36 @@ LxmfAdapter::PeerInfo* LxmfAdapter::rememberPeerIdentity(
{
copyCString(peer.display_name, sizeof(peer.display_name), display_name);
}
rtdir::LxmfAddressRecord directory_address{};
directory_address.valid = true;
copyHash(directory_address.destination_hash,
peer.destination_hash,
sizeof(directory_address.destination_hash));
copyHash(directory_address.identity_hash,
peer.identity_hash,
sizeof(directory_address.identity_hash));
memcpy(directory_address.enc_pub, peer.enc_pub, sizeof(directory_address.enc_pub));
memcpy(directory_address.sig_pub, peer.sig_pub, sizeof(directory_address.sig_pub));
copyCString(directory_address.display_name,
sizeof(directory_address.display_name),
peer.display_name);
directory_address.source = rtdir::EntrySource::RuntimeRx;
directory_address.first_seen_s = peer.last_seen_s;
directory_address.last_seen_s = peer.last_seen_s;
const auto address_status = rtdir::record_lxmf_address(directory_address);
if (address_status.sd_present && !address_status.saved)
const bool allow_persistence =
screen_runtime::is_sleeping() && !screen_runtime::is_saver_active();
if (allow_persistence)
{
Serial.printf("[LXMF][Directory] address_save failed message=%s detail=%s\n",
address_status.message,
address_status.detail);
rtdir::LxmfAddressRecord directory_address{};
directory_address.valid = true;
copyHash(directory_address.destination_hash,
peer.destination_hash,
sizeof(directory_address.destination_hash));
copyHash(directory_address.identity_hash,
peer.identity_hash,
sizeof(directory_address.identity_hash));
memcpy(directory_address.enc_pub, peer.enc_pub, sizeof(directory_address.enc_pub));
memcpy(directory_address.sig_pub, peer.sig_pub, sizeof(directory_address.sig_pub));
copyCString(directory_address.display_name,
sizeof(directory_address.display_name),
peer.display_name);
directory_address.source = rtdir::EntrySource::RuntimeRx;
directory_address.first_seen_s = peer.last_seen_s;
directory_address.last_seen_s = peer.last_seen_s;
const auto address_status = rtdir::record_lxmf_address(directory_address);
if (address_status.sd_present && !address_status.saved)
{
Serial.printf("[LXMF][Directory] address_save failed message=%s detail=%s\n",
address_status.message,
address_status.detail);
}
}
publishPeerUpdate(peer);
(void)maybePersistPeers(true);
(void)maybePersistPeers(false);
return &peer;
}
@@ -5697,7 +5894,7 @@ void LxmfAdapter::populateRxMeta(RxMeta* out) const
return;
}
const RxMeta& last_meta = interfaces_.lastRxMeta();
const RxMeta& last_meta = has_active_rx_meta_ ? active_rx_meta_ : interfaces_.lastRxMeta();
if (last_meta.rx_timestamp_ms != 0 || last_meta.rx_timestamp_s != 0)
{
*out = last_meta;
@@ -118,4 +118,9 @@ void ReticulumAdapter::setLastRxStats(float rssi, float snr)
service_->setLastRxStats(rssi, snr);
}
void ReticulumAdapter::processSendQueue()
{
service_->processSendQueue();
}
} // namespace chat::reticulum
@@ -0,0 +1,90 @@
#!/usr/bin/env python3
"""Guard Reticulum runtime budget boundaries from regressing."""
from __future__ import annotations
from pathlib import Path
import re
import sys
REPO_ROOT = Path(__file__).resolve().parent.parent
LXMF_CPP = REPO_ROOT / "platform/esp/arduino_common/src/chat/infra/lxmf/lxmf_adapter.cpp"
LXMF_H = REPO_ROOT / "platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/lxmf/lxmf_adapter.h"
RT_CPP = REPO_ROOT / "platform/esp/arduino_common/src/chat/infra/reticulum/reticulum_adapter.cpp"
RT_H = REPO_ROOT / "platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/reticulum/reticulum_adapter.h"
def method_body(text: str, signature: str) -> str:
start = text.find(signature)
if start < 0:
raise ValueError(f"missing method signature: {signature}")
brace = text.find("{", start)
if brace < 0:
raise ValueError(f"missing method body: {signature}")
depth = 0
for index in range(brace, len(text)):
char = text[index]
if char == "{":
depth += 1
elif char == "}":
depth -= 1
if depth == 0:
return text[brace + 1 : index]
raise ValueError(f"unterminated method body: {signature}")
def main() -> int:
violations: list[str] = []
lxmf_cpp = LXMF_CPP.read_text(encoding="utf-8")
lxmf_h = LXMF_H.read_text(encoding="utf-8")
rt_cpp = RT_CPP.read_text(encoding="utf-8")
rt_h = RT_H.read_text(encoding="utf-8")
for signature in (
"bool LxmfAdapter::pollIncomingText",
"bool LxmfAdapter::pollIncomingData",
):
body = method_body(lxmf_cpp, signature)
for forbidden in ("processRadioPackets", "processRuntime", "maybeAnnounce"):
if forbidden in body:
violations.append(f"{signature} must not call {forbidden}()")
if "maybePersistPeers(true)" in lxmf_cpp:
violations.append("Reticulum RX paths must not force maybePersistPeers(true)")
for stale_name in ("kWifiDiscoverySampleIntervalMs", "consumeWifiDiscoveryBudget"):
if stale_name in lxmf_cpp or stale_name in lxmf_h:
violations.append(f"stale Wi-Fi-only discovery budget name remains: {stale_name}")
if "void processSendQueue() override;" not in rt_h:
violations.append("ReticulumAdapter must expose processSendQueue() as the runtime pump hook")
if not re.search(r"void\s+ReticulumAdapter::processSendQueue\s*\(\s*\)\s*\{[^{}]*service_->processSendQueue\s*\(\s*\)", rt_cpp, re.S):
violations.append("ReticulumAdapter::processSendQueue() must forward to the LXMF service")
process_pos = lxmf_cpp.find("bool LxmfAdapter::processOneRadioPacket")
defer_pos = lxmf_cpp.find("shouldDeferDiscoveryPacket", process_pos)
remember_pos = lxmf_cpp.find("rememberPacket(packet_hash)", process_pos)
if process_pos < 0 or defer_pos < 0 or remember_pos < 0 or defer_pos > remember_pos:
violations.append("Reticulum packets must be eligible for discovery deferral before rememberPacket()")
if "record_announce(directory_announce)" in method_body(lxmf_cpp, "bool LxmfAdapter::handleAnnouncePacket"):
announce_body = method_body(lxmf_cpp, "bool LxmfAdapter::handleAnnouncePacket")
record_pos = announce_body.find("record_announce(directory_announce)")
gate_pos = announce_body.rfind("if (allow_persistence)", 0, record_pos)
if gate_pos < 0:
violations.append("record_announce() must be gated by allow_persistence")
if violations:
print("Reticulum runtime budget policy check failed:")
for violation in violations:
print(f"- {violation}")
return 1
print("Reticulum runtime budget policy check passed.")
return 0
if __name__ == "__main__":
sys.exit(main())