dandri/trail-mate
1
0
Fork 0
mirror of https://github.com/vicliu624/trail-mate.git synced 2026-08-06 18:59:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(runtime): stabilize storage and radio arbitration

Browse Source
This commit is contained in:
liu weikai
2026-07-21 08:02:13 +08:00
parent a7c1086be2
commit 4e291eed5b
30 changed files with 666 additions and 355 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/linux_sim_shell/CMakeLists.txt
+28 -3
View File
@@ -299,7 +299,7 @@ if(BUILD_TESTING)
"${TRAIL_MATE_REPO_ROOT}")
add_executable(trailmate_lxmf_runtime_budget_smoke
"${TRAIL_MATE_REPO_ROOT}/modules/core_chat/tests/test_lxmf_runtime_budget.cpp"
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/tests/test_lxmf_runtime_budget.cpp"
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/src/chat/infra/lxmf/lxmf_runtime_budget.cpp")
target_include_directories(trailmate_lxmf_runtime_budget_smoke
PRIVATE
@@ -311,7 +311,7 @@ if(BUILD_TESTING)
COMMAND trailmate_lxmf_runtime_budget_smoke)
add_executable(trailmate_lxmf_announce_scheduler_smoke
"${TRAIL_MATE_REPO_ROOT}/modules/core_chat/tests/test_lxmf_announce_scheduler.cpp"
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/tests/test_lxmf_announce_scheduler.cpp"
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/src/chat/infra/lxmf/lxmf_announce_scheduler.cpp")
target_include_directories(trailmate_lxmf_announce_scheduler_smoke
PRIVATE
@@ -322,7 +322,7 @@ if(BUILD_TESTING)
COMMAND trailmate_lxmf_announce_scheduler_smoke)
add_executable(trailmate_lxmf_deferred_discovery_queue_smoke
"${TRAIL_MATE_REPO_ROOT}/modules/core_chat/tests/test_lxmf_deferred_discovery_queue.cpp"
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/tests/test_lxmf_deferred_discovery_queue.cpp"
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/src/chat/infra/lxmf/lxmf_deferred_discovery_queue.cpp")
target_include_directories(trailmate_lxmf_deferred_discovery_queue_smoke
PRIVATE
@@ -335,6 +335,31 @@ if(BUILD_TESTING)
add_test(NAME trailmate_lxmf_deferred_discovery_queue_smoke
COMMAND trailmate_lxmf_deferred_discovery_queue_smoke)
add_executable(trailmate_protocol_chat_storage_v2_codec_smoke
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/tests/test_protocol_chat_storage_v2_codec.cpp"
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/src/chat/infra/store/protocol_chat_codec.cpp")
target_include_directories(trailmate_protocol_chat_storage_v2_codec_smoke
PRIVATE
"${TRAIL_MATE_REPO_ROOT}/modules/core_chat/include"
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/include")
target_compile_features(trailmate_protocol_chat_storage_v2_codec_smoke
PRIVATE cxx_std_17)
add_test(NAME trailmate_protocol_chat_storage_v2_codec_smoke
COMMAND trailmate_protocol_chat_storage_v2_codec_smoke)
add_executable(trailmate_protocol_peer_storage_v2_codec_smoke
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/tests/test_protocol_peer_storage_v2_codec.cpp"
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/src/chat/infra/store/protocol_peer_codec.cpp"
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/src/chat/infra/store/protocol_chat_codec.cpp")
target_include_directories(trailmate_protocol_peer_storage_v2_codec_smoke
PRIVATE
"${TRAIL_MATE_REPO_ROOT}/modules/core_chat/include"
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/include")
target_compile_features(trailmate_protocol_peer_storage_v2_codec_smoke
PRIVATE cxx_std_17)
add_test(NAME trailmate_protocol_peer_storage_v2_codec_smoke
COMMAND trailmate_protocol_peer_storage_v2_codec_smoke)
add_executable(trailmate_chat_delivery_event_projection_adapter_smoke
"${TRAIL_MATE_REPO_ROOT}/modules/ui_chat_runtime/tests/test_chat_delivery_event_projection_adapter.cpp"
"${TRAIL_MATE_REPO_ROOT}/modules/ui_chat_runtime/src/chat_delivery_event_projection_adapter.cpp"
boards/tdeck/src/tdeck_board.cpp
+5 -2
View File
@@ -143,7 +143,10 @@ bool withSharedSpiRadioAccess(const char* owner,
request);
if (!bus_token.acquired())
{
log_radio_spi_lock_timeout(owner, wait_ticks);
if (wait_ticks != 0)
{
log_radio_spi_lock_timeout(owner, wait_ticks);
}
return false;
}
fn();
@@ -1121,7 +1124,7 @@ int TDeckBoard::startRadioReceive()
uint32_t TDeckBoard::getRadioIrqFlags()
{
uint32_t flags = 0;
if (withSharedSpiRadioAccess("radio_irq", pdMS_TO_TICKS(20), [&]()
if (withSharedSpiRadioAccess("radio_irq", 0, [&]()
{ flags = radio_.getIrqFlags(); }))
{
return flags;
boards/tlora_pager/src/tlora_pager_board.cpp
+5 -2
View File
@@ -216,7 +216,10 @@ bool withSharedSpiRadioAccess(const char* owner,
request);
if (!bus_token.acquired())
{
log_radio_spi_lock_timeout(owner, wait_ticks);
if (wait_ticks != 0)
{
log_radio_spi_lock_timeout(owner, wait_ticks);
}
return false;
}
fn();
@@ -1753,7 +1756,7 @@ int TLoRaPagerBoard::startRadioReceive()
uint32_t TLoRaPagerBoard::getRadioIrqFlags()
{
uint32_t flags = 0;
if (withSharedSpiRadioAccess("radio_irq", pdMS_TO_TICKS(20), [&]()
if (withSharedSpiRadioAccess("radio_irq", 0, [&]()
{ flags = radio_.getIrqFlags(); }))
{
return flags;
docs/design/PROTOCOL_PARTITIONED_STORAGE_V2_DETAILED_DESIGN.md
+31 -14
View File
@@ -81,7 +81,7 @@ repository. They have no files and no independent caches.
```text
create SdStore
-> acquire startup storage lease
-> acquire store aggregate mutex
-> recover snapshot backups
-> load protocol projections
-> reconcile catalog from message segments
@@ -89,7 +89,7 @@ create SdStore
-> compact threshold-exceeding projections
create SdProtocolPeerRepository(SdStore)
-> acquire startup storage lease
-> acquire peer aggregate mutex
-> create /data/v2/{mt,mc,rt}
-> recover peer/contact snapshot backups
-> load peer snapshot + delta
@@ -153,7 +153,7 @@ the tombstone is durable.
1. Resolve the node ID in the active protocol partition.
2. Require MT node identity, MC public key, or RT destination identity.
3. Build a complete contact projection with alias and flags.
4. Acquire a bounded durable storage lease.
4. Hold the peer aggregate mutex while constructing the complete transaction.
5. Append and flush `contacts.delta`.
6. Apply the projection to memory.
7. Refresh only the affected peer/node projection.
@@ -181,7 +181,8 @@ eligible peer exists, insertion returns `CapacityExceeded`.
## Pending Peer Delta Queue
Peer observations must not block packet or UI work on shared SPI contention.
Peer observations use bounded per-operation SD access and must not own the
physical SPI bus across a repository transaction.
The queue uses a PSRAM vector plus a head index; draining does not repeatedly
erase the first vector element. Consecutive pending full-record updates for the
@@ -201,8 +202,9 @@ delta thresholds are exceeded:
- peer/contact: 1024 deltas.
Normal chat `flush()` does not inspect and compact all healthy journals. It
only retries a dirty protocol projection, at most once per five seconds, using
a non-blocking background SPI lease.
only retries a dirty protocol projection, at most once per five seconds. Each
filesystem operation uses the bounded SD runtime guard; compaction never wraps
the complete operation sequence in a physical SPI lease.
Snapshot replacement sequence:
@@ -236,15 +238,29 @@ through backup files.
## Concurrency And SPI
The peer repository serializes its aggregate with a FreeRTOS mutex. It holds one
repository lock while applying an aggregate transaction.
`SdStore` and `SdProtocolPeerRepository` each own a recursive FreeRTOS mutex.
The mutex serializes in-memory vectors, journal sequence numbers, merge
invariants, and the logical transaction that connects them. Recursive locking
is required because public query/update methods may call another method on the
same owner. This mutex is a state lock, not a bus lock.
All peer/contact journal I/O uses the shared storage bus arbiter. Peer appends
request a zero-wait background lease; contact edits use a bounded durable lease.
`sd_card_runtime` is the only normal owner of physical SD/display/radio SPI
arbitration. `open`, `read`, `write`, `flush`, `exists`, `rename`, and `remove`
take bounded per-operation guards. Store/repository/page-cache code must not
acquire `PersistenceBusGate` or `SharedSpiBusAdapter` around a sequence of those
operations. Doing so stretches physical ownership across CPU work and creates
radio/display starvation even when every nested file call is individually
bounded.
Chat authoritative append uses a zero-wait durable lease and delegates retry to
`MessageLedger`. No renderer waits for radio TX, MQTT forwarding, LoRa airtime,
or projection compaction.
Explicit hardware sessions are separate: SD unmount/recovery, USB mass storage,
and user-visible external font loading may own an exclusive bus/session token
under their dedicated lifecycle specification. They must not be copied into a
normal message, peer, or page-cache repository.
No renderer waits for radio TX, MQTT forwarding, LoRa airtime, or projection
compaction. SPI contention can delay or fail a bounded filesystem operation;
the ledger/repository retry policy handles that result without converting a
state mutex into physical bus ownership.
## Test Matrix
@@ -255,6 +271,8 @@ or projection compaction.
| Seen journal corrupt | Rebuilt from RT message journals |
| Peer append SPI busy | Peer visible in memory; ordered delta queued |
| Contact append SPI busy | Contact edit fails; memory unchanged |
| Chat append while radio IRQ polling | SD and radio take separate bounded physical turns; no transaction-wide SPI hold |
| Nomad page cache read during LoRa activity | Page state is serialized; SD runtime releases SPI between file operations |
| MC node observed before key | Temporary peer persists; cannot become contact |
| MC key later arrives | Stable record then temporary tombstone |
| RT protocol event before destination identity | Temporary record upgrades to destination |
@@ -262,4 +280,3 @@ or projection compaction.
| Nearby capacity full | Oldest unprotected nearby evicted |
| All peers protected | New nearby rejected; contacts retained |
| Power loss during snapshot rename | Backup restored at startup |
docs/design/PROTOCOL_PARTITIONED_STORAGE_V2_OVERVIEW.md
+6
View File
@@ -43,6 +43,12 @@ projections, not authoritative mutable blobs.
**CQRS projection:** UI queries read catalog/contact/node projections. Incoming
message and peer writes do not wait for expensive UI-oriented reconstruction.
**Two-level concurrency ownership:** Store/repository recursive mutexes protect
aggregate state and sequence invariants. They never represent physical SPI
ownership. `SdRuntimeFile`/`SdRuntimeDir` alone acquire the shared SPI bus for
each bounded filesystem operation. A repository transaction must not hold a
shared-SPI token around a sequence of SD runtime calls.
## High-Level Commit Flows
### Incoming Message
docs/protocol_runtime_budget_policy.md
+28
View File
@@ -28,6 +28,14 @@ longer ready. In `WifiGatewayOnly`, shared LoRa RX stays suppressed. In
`LoRaOnly`, shared LoRa RX stays enabled. Outbound LoRa TX may still be queued
only when the active strategy has selected LoRa.
Carrier selection is a runtime fact, distinct from configured interfaces.
`sendPacket`, `sendPacketOn`, packet polling, legacy LoRa polling, raw LoRa
ingress, and the shared RX gate must consult the same selection owner. A path
learned on a previously selected interface must not reactivate that interface;
when its interface is no longer selected, routing falls back to the original
header on the currently selected carrier so that path discovery can converge
there.
When the screen is on, realtime processing is limited to:
- LXMF direct traffic addressed to this device.
@@ -182,6 +190,26 @@ Forbidden scheduler shapes:
- Separate local drain counters that allow protocol actions, app sends, ACK
retry, and MQTT downlink each to consume a full TX slot in the same tick.
Shared-SPI busy is a deferred radio condition, not successful airtime. The
radio task retains the front TX packet and retries it with bounded exponential
backoff when the board reports SPI access busy. IRQ polling uses a zero-wait
probe and must never block a frame-critical display operation. Queue buffers and
RX scratch use PSRAM on PSRAM-capable targets.
## Protocol Switch Lifecycle
Changing MT/MC/RT is a runtime lifecycle transition, not a reboot contract.
Before a backend is configured or installed, the radio and mesh tasks must
cooperatively reach quiescent points outside board SPI calls and adapter work.
Only then may the owner discard old-protocol TX/RX queues, configure the radio,
install the backend, and switch Chat/Contacts active protocol projections.
If quiescing or installation fails, the old protocol remains active and its
configuration is reapplied. A task must never be force-suspended while it may
hold a shared-SPI lock. Settings reports success/failure from this transition;
it must not issue an unconditional software reset that makes a successful
switch look like a crash.
## Reticulum Runtime Owner Budget
The embedded Reticulum adapter is allowed to coordinate owners, but it must not
modules/core_chat/src/infra/mesh_peer_directory_core.cpp
+14 -10
View File
@@ -740,7 +740,8 @@ MeshPeerDirectoryStatus MeshPeerDirectoryCore::search(
MeshPeerDirectoryStatusCode::InvalidArgument);
}
*out_count = 0;
std::vector<const MeshPeerRecord*> matches;
matches.reserve(records_.size());
for (const auto& record : records_)
{
if (!meshPeerSameProtocol(record.identity.protocol, protocol) ||
@@ -748,15 +749,18 @@ MeshPeerDirectoryStatus MeshPeerDirectoryCore::search(
{
continue;
}
if (*out_count < max_records)
{
out_records[*out_count] = record;
}
++(*out_count);
if (*out_count >= max_records)
{
break;
}
matches.push_back(&record);
}
std::sort(matches.begin(),
matches.end(),
[](const MeshPeerRecord* lhs, const MeshPeerRecord* rhs)
{
return lhs->last_seen_s > rhs->last_seen_s;
});
*out_count = std::min(max_records, matches.size());
for (std::size_t index = 0; index < *out_count; ++index)
{
out_records[index] = *matches[index];
}
return MeshPeerDirectoryStatus::success();
}
modules/core_chat/tests/test_mesh_peer_directory_contract.cpp
+21 -14
View File
@@ -171,7 +171,7 @@ class MemoryMeshPeerDirectory final : public chat::IMeshPeerDirectory
chat::MeshPeerDirectoryStatusCode::InvalidArgument);
}
*out_count = 0;
std::vector<chat::MeshPeerRecord> matches;
for (const auto& record : records_)
{
if (!chat::meshPeerSameProtocol(record.identity.protocol, protocol))
@@ -182,15 +182,19 @@ class MemoryMeshPeerDirectory final : public chat::IMeshPeerDirectory
{
continue;
}
if (*out_count < max_records)
{
out_records[*out_count] = record;
}
++(*out_count);
if (*out_count >= max_records)
{
break;
}
matches.push_back(record);
}
std::sort(matches.begin(),
matches.end(),
[](const chat::MeshPeerRecord& lhs,
const chat::MeshPeerRecord& rhs)
{
return lhs.last_seen_s > rhs.last_seen_s;
});
*out_count = std::min(max_records, matches.size());
for (std::size_t index = 0; index < *out_count; ++index)
{
out_records[index] = matches[index];
}
return chat::MeshPeerDirectoryStatus::success();
}
@@ -640,21 +644,24 @@ void search_and_user_flags_are_directory_behaviors()
assert(directory.begin().succeeded());
auto beta = makeReticulumPeer(0x60, "beta trail", 1);
auto recent_beta = makeReticulumPeer(0x62, "beta summit", 3);
assert(directory.record(makeReticulumPeer(0x61, "alpha trail", 2))
.succeeded());
assert(directory.record(beta).succeeded());
assert(directory.record(recent_beta).succeeded());
chat::MeshPeerRecord records[1]{};
chat::MeshPeerRecord records[2]{};
std::size_t count = 0;
assert(directory
.search(chat::MeshProtocol::Reticulum,
"beta",
records,
1,
2,
&count)
.succeeded());
assert(count == 1);
assert(std::strcmp(records[0].display_name, "beta trail") == 0);
assert(count == 2);
assert(std::strcmp(records[0].display_name, "beta summit") == 0);
assert(std::strcmp(records[1].display_name, "beta trail") == 0);
chat::MeshPeerUserFlags flags{};
flags.favorite = true;
modules/ui_shared/src/ui/screens/contacts/contacts_page_runtime.cpp
+14
View File
@@ -593,6 +593,20 @@ void refresh_contacts_data_impl_internal()
filter_to_active_protocol(g_contacts_state.nearby_list);
filter_to_active_protocol(g_contacts_state.ignored_list);
merge_reticulum_directory_projection();
const chat::MeshProtocol active_protocol =
chat::infra::normalizeMeshProtocol(
app::configFacade().getConfig().mesh_protocol);
if (chat::infra::isReticulumMeshProtocol(active_protocol))
{
std::stable_sort(
g_contacts_state.nearby_list.begin(),
g_contacts_state.nearby_list.end(),
[](const chat::contacts::NodeInfo& lhs,
const chat::contacts::NodeInfo& rhs)
{
return lhs.last_seen > rhs.last_seen;
});
}
mark_contacts_data_refreshed();
CONTACTS_LOG("[Contacts] Data refreshed: %zu contacts, %zu nearby, %zu groups, %zu ignored\n",
modules/ui_shared/src/ui/screens/settings/settings_page_components.cpp
-1
View File
@@ -3262,7 +3262,6 @@ static void on_option_clicked(lv_event_t* e)
{
rebuild_list = true;
::ui::feedback::show_notice(::ui::i18n::tr("Protocol switched"), 2000);
restart_now = true;
}
}
if (id == settings::ui::SettingId::ChatRegion)
platform/esp/arduino_common/include/platform/esp/arduino_common/app_tasks.h
+11 -5
View File
@@ -27,11 +27,14 @@ class AppTasks
struct RadioPacket
{
uint8_t* data;
size_t size;
bool is_tx; // true for TX, false for RX
uint8_t* data = nullptr;
size_t size = 0;
bool is_tx = false; // true for TX, false for RX
float rssi = 0.0f;
float snr = 0.0f;
uint32_t queued_ms = 0;
uint32_t next_attempt_ms = 0;
uint8_t retry_count = 0;
};
/**
@@ -80,7 +83,7 @@ class AppTasks
/**
* @brief Pause radio + mesh tasks (for exclusive radio modes like walkie-talkie)
*/
static void pauseRadioTasks();
static bool pauseRadioTasks(uint32_t timeout_ms = 2000U);
/**
* @brief Resume radio + mesh tasks after pause
@@ -130,11 +133,14 @@ class AppTasks
static TaskHandle_t mesh_task_handle_;
static LoraBoard* board_;
static chat::IMeshAdapter* adapter_;
static bool radio_tasks_paused_;
static uint8_t* radio_rx_scratch_;
static volatile bool radio_tasks_paused_;
static volatile bool radio_receive_active_;
static volatile bool radio_receive_restart_pending_;
static volatile bool radio_receive_suppressed_;
static volatile bool radio_transmit_active_;
static volatile bool radio_task_quiesced_;
static volatile bool mesh_task_quiesced_;
};
} // namespace app
platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/reticulum/reticulum_interfaces.h
+5 -1
View File
@@ -268,6 +268,7 @@ class ReticulumInterfaceSet
bool hasReadyInterface() const;
bool hasReadyWifiGateway() const;
bool wifiGatewayConfigured() const;
bool isInterfaceSelected(InterfaceId interface_id) const;
bool sendPacket(const uint8_t* data, size_t len);
bool sendPacketOn(InterfaceId interface_id,
const uint8_t* data,
@@ -303,7 +304,10 @@ class ReticulumInterfaceSet
uint8_t next_poll_index_ = 0;
bool shared_lora_rx_suppressed_ = false;
bool loraEnabled() const;
bool loraAllowed() const;
bool wifiAllowed() const;
bool loraSelectedForRuntime() const;
bool wifiSelectedForRuntime() const;
bool hasConfiguredIpInterface() const;
void syncSharedLoRaRxGate();
};
platform/esp/arduino_common/include/platform/esp/arduino_common/chat/infra/store/sd_store.h
+5 -1
View File
@@ -10,6 +10,9 @@
#include "platform/esp/arduino_common/chat/infra/store/protocol_chat_codec.h"
#include "platform/esp/arduino_common/memory/psram_allocator.h"
#include "freertos/FreeRTOS.h"
#include "freertos/semphr.h"
#include <cstddef>
#include <cstdint>
#include <vector>
@@ -30,7 +33,7 @@ class SdStore final : public IChatStore
static constexpr std::size_t kSeenHotCapacity = 4096;
SdStore();
~SdStore() override = default;
~SdStore() override;
bool isReady() const { return ready_; }
@@ -204,6 +207,7 @@ class SdStore final : public IChatStore
static bool removeTree(const char* path);
storage::v2::FixedSlotJournalEngine journal_{};
mutable SemaphoreHandle_t mutex_ = nullptr;
CatalogList catalog_{};
ReadStateList read_state_{};
StatusList statuses_{};
platform/esp/arduino_common/include/platform/esp/arduino_common/storage/scoped_state_lock.h
+40
View File
@@ -0,0 +1,40 @@
#pragma once
#include "freertos/FreeRTOS.h"
#include "freertos/semphr.h"
namespace platform::esp::arduino_common::storage
{
class ScopedRecursiveStateLock final
{
public:
explicit ScopedRecursiveStateLock(SemaphoreHandle_t mutex)
: mutex_(mutex),
locked_(mutex_ &&
xSemaphoreTakeRecursive(mutex_, portMAX_DELAY) == pdTRUE)
{
}
~ScopedRecursiveStateLock()
{
if (locked_)
{
xSemaphoreGiveRecursive(mutex_);
}
}
ScopedRecursiveStateLock(const ScopedRecursiveStateLock&) = delete;
ScopedRecursiveStateLock& operator=(const ScopedRecursiveStateLock&) = delete;
bool locked() const
{
return locked_;
}
private:
SemaphoreHandle_t mutex_ = nullptr;
bool locked_ = false;
};
} // namespace platform::esp::arduino_common::storage
platform/esp/arduino_common/src/app_context.cpp
+8
View File
@@ -565,6 +565,10 @@ bool AppContext::switchMeshProtocol(chat::MeshProtocol protocol, bool persist)
{
return false;
}
if (!AppTasks::pauseRadioTasks())
{
return false;
}
const chat::MeshProtocol previous_protocol = config_.mesh_protocol;
config_.mesh_protocol = normalized;
@@ -587,6 +591,9 @@ bool AppContext::switchMeshProtocol(chat::MeshProtocol protocol, bool persist)
if (!mesh_router_->installBackend(normalized, std::move(backend)))
{
config_.mesh_protocol = previous_protocol;
sync_reticulum_group_config(config_);
mesh_router_->applyConfig(config_.activeMeshConfig());
AppTasks::resumeRadioTasks();
return false;
}
@@ -603,6 +610,7 @@ bool AppContext::switchMeshProtocol(chat::MeshProtocol protocol, bool persist)
{
saveConfig();
}
AppTasks::resumeRadioTasks();
return true;
}
platform/esp/arduino_common/src/app_tasks.cpp
+189 -33
View File
@@ -17,6 +17,7 @@
#include <cstdio>
#define RADIOLIB_ERR_NONE 0
#define RADIOLIB_ERR_SPI_WRITE_FAILED (-16)
#define RADIOLIB_SX126X_IRQ_RX_DONE 0x0002U
#define RADIOLIB_SX126X_IRQ_HEADER_ERR 0x0020U
#define RADIOLIB_SX126X_IRQ_CRC_ERR 0x0040U
@@ -38,6 +39,7 @@ struct IdfSerialLogAdapter
static constexpr IdfSerialLogAdapter Serial{};
#endif
#include <algorithm>
#include <cstring>
#include <esp_heap_caps.h>
@@ -68,6 +70,8 @@ constexpr uint32_t kMeshTaskStackBytes = 8 * 1024;
constexpr uint32_t kRadioRxSummaryIntervalMs = 5000;
constexpr uint32_t kRadioForegroundRxSummaryIntervalMs = 15000;
constexpr uint32_t kRadioForegroundPostRxQuietMs = 120;
constexpr uint32_t kRadioTxRetryInitialMs = 25;
constexpr uint32_t kRadioTxRetryMaxMs = 400;
struct RadioRxSummary
{
@@ -110,6 +114,49 @@ bool radio_rx_quiet_window_active(uint32_t now_ms, uint32_t quiet_until_ms)
static_cast<int32_t>(quiet_until_ms - now_ms) > 0;
}
bool deadline_reached(uint32_t now_ms, uint32_t deadline_ms)
{
return deadline_ms == 0U ||
static_cast<int32_t>(now_ms - deadline_ms) >= 0;
}
uint32_t radio_tx_retry_delay_ms(uint8_t retry_count)
{
const uint8_t shift = retry_count > 4U ? 4U : retry_count;
const uint32_t delay_ms = kRadioTxRetryInitialMs << shift;
return std::min(delay_ms, kRadioTxRetryMaxMs);
}
bool transient_radio_tx_failure(int state)
{
return state == RADIOLIB_ERR_SPI_WRITE_FAILED;
}
void drain_radio_packet_queue(QueueHandle_t queue)
{
if (!queue)
{
return;
}
AppTasks::RadioPacket packet{};
while (xQueueReceive(queue, &packet, 0) == pdPASS)
{
if (packet.data)
{
heap_caps_free(packet.data);
}
}
}
uint8_t* allocate_radio_packet_buffer(std::size_t size)
{
const uint32_t caps =
heap_caps_get_total_size(MALLOC_CAP_SPIRAM) > 0
? MALLOC_CAP_SPIRAM | MALLOC_CAP_8BIT
: MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT;
return static_cast<uint8_t*>(heap_caps_malloc(size, caps));
}
uint32_t radio_rx_done_mask()
{
uint32_t mask = 0;
@@ -218,18 +265,33 @@ TaskHandle_t AppTasks::radio_task_handle_ = nullptr;
TaskHandle_t AppTasks::mesh_task_handle_ = nullptr;
LoraBoard* AppTasks::board_ = nullptr;
chat::IMeshAdapter* AppTasks::adapter_ = nullptr;
bool AppTasks::radio_tasks_paused_ = false;
uint8_t* AppTasks::radio_rx_scratch_ = nullptr;
volatile bool AppTasks::radio_tasks_paused_ = false;
volatile bool AppTasks::radio_receive_active_ = false;
volatile bool AppTasks::radio_receive_restart_pending_ = true;
volatile bool AppTasks::radio_receive_suppressed_ = false;
volatile bool AppTasks::radio_transmit_active_ = false;
volatile bool AppTasks::radio_task_quiesced_ = false;
volatile bool AppTasks::mesh_task_quiesced_ = false;
bool AppTasks::init(LoraBoard& board, chat::IMeshAdapter* adapter)
{
board_ = &board;
adapter_ = adapter;
radio_tasks_paused_ = false;
radio_task_quiesced_ = false;
mesh_task_quiesced_ = false;
radio_receive_active_ = false;
radio_receive_restart_pending_ = !radio_receive_suppressed_;
if (!radio_rx_scratch_)
{
radio_rx_scratch_ = allocate_radio_packet_buffer(255U);
}
if (!radio_rx_scratch_)
{
Serial.printf("[LORA] RX scratch allocation failed bytes=255\n");
return false;
}
// Create queues
radio_tx_queue_ = xQueueCreate(RADIO_QUEUE_SIZE, sizeof(RadioPacket));
@@ -267,36 +329,53 @@ bool AppTasks::init(LoraBoard& board, chat::IMeshAdapter* adapter)
return (result == pdPASS);
}
void AppTasks::pauseRadioTasks()
bool AppTasks::pauseRadioTasks(uint32_t timeout_ms)
{
if (radio_tasks_paused_)
{
return;
return (radio_task_handle_ == nullptr || radio_task_quiesced_) &&
(mesh_task_handle_ == nullptr || mesh_task_quiesced_);
}
const TaskHandle_t current_task = xTaskGetCurrentTaskHandle();
if (current_task == radio_task_handle_ || current_task == mesh_task_handle_)
{
Serial.printf("[LORA] task quiesce rejected caller=owned_task\n");
return false;
}
radio_tasks_paused_ = true;
requestRadioReceiveRestart();
if (radio_task_handle_)
const uint32_t start_ms = millis();
while ((radio_task_handle_ != nullptr && !radio_task_quiesced_) ||
(mesh_task_handle_ != nullptr && !mesh_task_quiesced_))
{
vTaskSuspend(radio_task_handle_);
}
if (mesh_task_handle_)
{
vTaskSuspend(mesh_task_handle_);
if (timeout_ms != 0U && millis() - start_ms >= timeout_ms)
{
radio_tasks_paused_ = false;
Serial.printf("[LORA] task quiesce timeout radio=%u mesh=%u wait_ms=%lu\n",
radio_task_quiesced_ ? 1U : 0U,
mesh_task_quiesced_ ? 1U : 0U,
static_cast<unsigned long>(millis() - start_ms));
return false;
}
vTaskDelay(pdMS_TO_TICKS(5));
}
if (radio_tx_queue_)
{
xQueueReset(radio_tx_queue_);
drain_radio_packet_queue(radio_tx_queue_);
}
if (radio_rx_queue_)
{
xQueueReset(radio_rx_queue_);
drain_radio_packet_queue(radio_rx_queue_);
}
if (mesh_queue_)
{
xQueueReset(mesh_queue_);
drain_radio_packet_queue(mesh_queue_);
}
Serial.printf("[LORA] tasks quiesced wait_ms=%lu\n",
static_cast<unsigned long>(millis() - start_ms));
return true;
}
void AppTasks::resumeRadioTasks()
@@ -307,14 +386,6 @@ void AppTasks::resumeRadioTasks()
}
radio_tasks_paused_ = false;
if (radio_task_handle_)
{
vTaskResume(radio_task_handle_);
}
if (mesh_task_handle_)
{
vTaskResume(mesh_task_handle_);
}
requestRadioReceiveRestart();
}
@@ -385,8 +456,7 @@ bool AppTasks::enqueueRadioTransmit(const uint8_t* data, size_t size)
return false;
}
uint8_t* copy = static_cast<uint8_t*>(
heap_caps_malloc(size, MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT));
uint8_t* copy = allocate_radio_packet_buffer(size);
if (!copy)
{
return false;
@@ -397,6 +467,7 @@ bool AppTasks::enqueueRadioTransmit(const uint8_t* data, size_t size)
packet.data = copy;
packet.size = size;
packet.is_tx = true;
packet.queued_ms = millis();
if (xQueueSend(radio_tx_queue_, &packet, 0) != pdPASS)
{
heap_caps_free(copy);
@@ -422,7 +493,6 @@ void AppTasks::radioTask(void* pvParameters)
{
(void)pvParameters;
uint8_t rx_buffer[255];
const uint32_t rx_done_mask = radio_rx_done_mask();
const uint32_t terminal_irq_mask = radio_terminal_irq_mask();
RadioRxSummary rx_summary{};
@@ -430,6 +500,14 @@ void AppTasks::radioTask(void* pvParameters)
while (true)
{
if (radio_tasks_paused_)
{
radio_task_quiesced_ = true;
vTaskDelay(pdMS_TO_TICKS(5));
continue;
}
radio_task_quiesced_ = false;
if (::platform::ui::reticulum_call::realtime_mode_active())
{
requestRadioReceiveRestart();
@@ -439,6 +517,7 @@ void AppTasks::radioTask(void* pvParameters)
bool should_restart_rx = radio_receive_restart_pending_;
bool handled_tx = false;
bool deferred_tx = false;
if (radio_transmit_active_)
{
@@ -447,12 +526,19 @@ void AppTasks::radioTask(void* pvParameters)
}
// Process TX queue
RadioPacket tx_packet;
if (xQueueReceive(radio_tx_queue_, &tx_packet, 0) == pdPASS)
RadioPacket tx_packet{};
const uint32_t tx_now_ms = millis();
RadioPacket queued_packet{};
const bool tx_ready =
xQueuePeek(radio_tx_queue_, &queued_packet, 0) == pdPASS &&
deadline_reached(tx_now_ms, queued_packet.next_attempt_ms);
if (tx_ready &&
xQueueReceive(radio_tx_queue_, &tx_packet, 0) == pdPASS)
{
handled_tx = tx_packet.is_tx && tx_packet.data && tx_packet.size > 0;
if (tx_packet.is_tx && tx_packet.data && tx_packet.size > 0)
{
bool keep_packet = false;
// Send packet
if (board_ && board_->isRadioOnline())
{
@@ -462,6 +548,55 @@ void AppTasks::radioTask(void* pvParameters)
state = board_->transmitRadio(tx_packet.data, tx_packet.size);
setRadioTransmitActive(false);
LORA_LOG("[LORA] TX queue len=%u state=%d\n", (unsigned)tx_packet.size, state);
if (transient_radio_tx_failure(state))
{
if (tx_packet.retry_count != UINT8_MAX)
{
++tx_packet.retry_count;
}
tx_packet.next_attempt_ms =
millis() + radio_tx_retry_delay_ms(tx_packet.retry_count);
if (xQueueSendToFront(radio_tx_queue_, &tx_packet, 0) == pdPASS)
{
keep_packet = true;
deferred_tx = true;
LORA_LOG("[LORA] TX deferred len=%u state=%d retry=%u age_ms=%lu\n",
static_cast<unsigned>(tx_packet.size),
state,
static_cast<unsigned>(tx_packet.retry_count),
static_cast<unsigned long>(millis() - tx_packet.queued_ms));
if (tx_packet.retry_count == 1U ||
(tx_packet.retry_count % 16U) == 0U)
{
Serial.printf("[LORA] TX deferred reason=spi_busy len=%u retry=%u age_ms=%lu\n",
static_cast<unsigned>(tx_packet.size),
static_cast<unsigned>(tx_packet.retry_count),
static_cast<unsigned long>(millis() - tx_packet.queued_ms));
}
}
else
{
Serial.printf("[LORA] TX retry queue lost len=%u state=%d retry=%u\n",
static_cast<unsigned>(tx_packet.size),
state,
static_cast<unsigned>(tx_packet.retry_count));
}
}
else if (state == RADIOLIB_ERR_NONE &&
tx_packet.retry_count != 0U)
{
Serial.printf("[LORA] TX recovered len=%u retries=%u age_ms=%lu\n",
static_cast<unsigned>(tx_packet.size),
static_cast<unsigned>(tx_packet.retry_count),
static_cast<unsigned long>(millis() - tx_packet.queued_ms));
}
else if (state != RADIOLIB_ERR_NONE)
{
Serial.printf("[LORA] TX failed len=%u state=%d age_ms=%lu\n",
static_cast<unsigned>(tx_packet.size),
state,
static_cast<unsigned long>(millis() - tx_packet.queued_ms));
}
if (state == RADIOLIB_ERR_NONE && !radio_receive_suppressed_)
{
int rx_state = board_->startRadioReceive();
@@ -490,10 +625,20 @@ void AppTasks::radioTask(void* pvParameters)
{
LORA_LOG("[LORA] TX drop (radio offline) len=%u\n", (unsigned)tx_packet.size);
}
free(tx_packet.data);
if (!keep_packet)
{
heap_caps_free(tx_packet.data);
}
}
}
if (deferred_tx)
{
maybe_log_radio_rx_summary(rx_summary);
vTaskDelay(kRadioDisplayPressurePollDelay);
continue;
}
if (radio_receive_suppressed_)
{
radio_receive_active_ = false;
@@ -548,14 +693,17 @@ void AppTasks::radioTask(void* pvParameters)
packet_length = static_cast<int>(board_->getRadioPacketLength(true));
if (packet_length > 0 && packet_length <= 255)
{
int state = board_->readRadioData(rx_buffer, packet_length);
int state = board_->readRadioData(radio_rx_scratch_, packet_length);
if (radio_read_state_has_payload(state))
{
RadioPacket rx_packet;
rx_packet.data = (uint8_t*)malloc(packet_length);
rx_packet.data =
allocate_radio_packet_buffer(packet_length);
if (rx_packet.data)
{
memcpy(rx_packet.data, rx_buffer, packet_length);
memcpy(rx_packet.data,
radio_rx_scratch_,
packet_length);
rx_packet.size = packet_length;
rx_packet.is_tx = false;
rx_packet.rssi = board_->getRadioRSSI();
@@ -565,7 +713,7 @@ void AppTasks::radioTask(void* pvParameters)
rx_summary.bytes += static_cast<uint32_t>(packet_length);
if (xQueueSend(mesh_queue_, &rx_packet, 0) != pdPASS)
{
free(rx_packet.data);
heap_caps_free(rx_packet.data);
++rx_summary.queue_drops;
}
}
@@ -626,6 +774,14 @@ void AppTasks::meshTask(void* pvParameters)
while (true)
{
if (radio_tasks_paused_)
{
mesh_task_quiesced_ = true;
vTaskDelay(pdMS_TO_TICKS(5));
continue;
}
mesh_task_quiesced_ = false;
// Process received packets
RadioPacket rx_packet;
if (::platform::ui::reticulum_call::realtime_mode_active())
@@ -634,7 +790,7 @@ void AppTasks::meshTask(void* pvParameters)
{
if (rx_packet.data)
{
free(rx_packet.data);
heap_caps_free(rx_packet.data);
}
}
if (adapter_)
@@ -653,7 +809,7 @@ void AppTasks::meshTask(void* pvParameters)
adapter_->handleRawPacket(rx_packet.data, rx_packet.size);
// Free buffer
free(rx_packet.data);
heap_caps_free(rx_packet.data);
}
}
platform/esp/arduino_common/src/chat/infra/lxmf/lxmf_adapter.cpp
+4
View File
@@ -7233,6 +7233,10 @@ bool LxmfAdapter::routeAndSendPacket(const uint8_t* raw_packet, size_t raw_len,
{
return send_packet(raw_packet, raw_len);
}
if (!interfaces_.isInterfaceSelected(path->interface_id))
{
return send_packet(raw_packet, raw_len);
}
if (path->hops <= 1 || path->direct)
{
return send_packet(raw_packet, raw_len, path->interface_id);
platform/esp/arduino_common/src/chat/infra/reticulum/reticulum_interfaces.cpp
+85 -18
View File
@@ -1670,11 +1670,8 @@ void ReticulumInterfaceSet::maintain()
bool ReticulumInterfaceSet::hasReadyInterface() const
{
if (loraEnabled() && lora_.isReady())
{
return true;
}
return hasReadyWifiGateway();
return (loraSelectedForRuntime() && lora_.isReady()) ||
(wifiSelectedForRuntime() && hasReadyWifiGateway());
}
bool ReticulumInterfaceSet::hasReadyWifiGateway() const
@@ -1698,6 +1695,30 @@ bool ReticulumInterfaceSet::wifiGatewayConfigured() const
return hasConfiguredIpInterface();
}
bool ReticulumInterfaceSet::isInterfaceSelected(InterfaceId interface_id) const
{
if (interface_id == kLoRaInterfaceId)
{
return loraSelectedForRuntime();
}
if (!wifiSelectedForRuntime())
{
return false;
}
if (auto_.owns(interface_id))
{
return auto_.isReady();
}
for (uint8_t index = 0; index < tcp_count_; ++index)
{
if (tcp_[index].interfaceId() == interface_id)
{
return tcp_[index].isReady();
}
}
return false;
}
bool ReticulumInterfaceSet::sendPacket(const uint8_t* data, size_t len)
{
last_tx_result_ = {};
@@ -1707,25 +1728,28 @@ bool ReticulumInterfaceSet::sendPacket(const uint8_t* data, size_t len)
}
maintain();
last_tx_result_.lora_required = loraEnabled();
last_tx_result_.lora_required = loraSelectedForRuntime();
last_tx_result_.lora_ready =
last_tx_result_.lora_required && lora_.isReady();
last_tx_result_.wifi_required = hasConfiguredIpInterface();
last_tx_result_.wifi_ready = hasReadyWifiGateway();
last_tx_result_.wifi_required = wifiSelectedForRuntime();
last_tx_result_.wifi_ready =
last_tx_result_.wifi_required && hasReadyWifiGateway();
if (last_tx_result_.lora_ready && lora_.sendPacket(data, len))
{
last_tx_result_.lora_ok = true;
++last_tx_result_.sent_count;
}
if (auto_.isReady() && auto_.sendPacket(data, len))
if (last_tx_result_.wifi_required && auto_.isReady() &&
auto_.sendPacket(data, len))
{
last_tx_result_.wifi_ok = true;
++last_tx_result_.sent_count;
}
for (uint8_t index = 0; index < tcp_count_; ++index)
{
if (tcp_[index].isReady() && tcp_[index].sendPacket(data, len))
if (last_tx_result_.wifi_required && tcp_[index].isReady() &&
tcp_[index].sendPacket(data, len))
{
last_tx_result_.wifi_ok = true;
++last_tx_result_.sent_count;
@@ -1765,13 +1789,19 @@ bool ReticulumInterfaceSet::sendPacketOn(InterfaceId interface_id,
if (interface_id == kLoRaInterfaceId)
{
last_tx_result_.lora_required = true;
last_tx_result_.lora_ready = loraEnabled() && lora_.isReady();
last_tx_result_.lora_ready =
loraSelectedForRuntime() && lora_.isReady();
sent = last_tx_result_.lora_ready && lora_.sendPacket(data, len);
last_tx_result_.lora_ok = sent;
}
else
{
last_tx_result_.wifi_required = true;
const bool call_wifi_override = call_link_id != nullptr;
if (!call_wifi_override && !wifiSelectedForRuntime())
{
return false;
}
if (auto_.owns(interface_id))
{
last_tx_result_.wifi_ready = auto_.isReady();
@@ -1863,6 +1893,8 @@ bool ReticulumInterfaceSet::pollIncomingPacket(RxPacket* out)
}
maintain();
const bool lora_selected = loraSelectedForRuntime();
const bool wifi_selected = wifiSelectedForRuntime();
const uint8_t source_count = static_cast<uint8_t>(2U + tcp_count_);
for (uint8_t offset = 0; offset < source_count; ++offset)
{
@@ -1871,15 +1903,15 @@ bool ReticulumInterfaceSet::pollIncomingPacket(RxPacket* out)
bool got = false;
if (index == 0)
{
got = loraEnabled() && lora_.pollPacket(out);
got = lora_selected && lora_.pollPacket(out);
}
else if (index == 1)
{
got = auto_.pollPacket(out);
got = wifi_selected && auto_.pollPacket(out);
}
else
{
got = tcp_[index - 2U].pollPacket(out);
got = wifi_selected && tcp_[index - 2U].pollPacket(out);
}
if (got)
{
@@ -1895,12 +1927,12 @@ bool ReticulumInterfaceSet::pollIncomingPacket(RxPacket* out)
bool ReticulumInterfaceSet::pollLegacyIncomingData(MeshIncomingData* out)
{
return loraEnabled() && lora_.pollLegacyIncomingData(out);
return loraSelectedForRuntime() && lora_.pollLegacyIncomingData(out);
}
void ReticulumInterfaceSet::handleRawPacket(const uint8_t* data, size_t size)
{
if (loraEnabled())
if (loraSelectedForRuntime())
{
lora_.handleRawPacket(data, size);
}
@@ -1929,12 +1961,18 @@ float ReticulumInterfaceSet::lastRxSnr() const
return lora_.lastRxSnr();
}
bool ReticulumInterfaceSet::loraEnabled() const
bool ReticulumInterfaceSet::loraAllowed() const
{
if (::platform::ui::reticulum_call::resource_preempt_active())
{
return false;
}
if (!config_.reticulum_lora_enabled ||
config_.reticulum_interface_policy ==
ReticulumInterfacePolicy::WifiGatewayOnly)
{
return false;
}
const size_t interface_count = std::min<size_t>(
network_config_.interface_count,
reticulum::kMaxNetworkInterfaces);
@@ -1951,6 +1989,35 @@ bool ReticulumInterfaceSet::loraEnabled() const
return false;
}
bool ReticulumInterfaceSet::wifiAllowed() const
{
return config_.reticulum_wifi_gateway_enabled &&
config_.reticulum_interface_policy !=
ReticulumInterfacePolicy::LoRaOnly &&
hasConfiguredIpInterface();
}
bool ReticulumInterfaceSet::loraSelectedForRuntime() const
{
if (!loraAllowed())
{
return false;
}
return config_.reticulum_interface_policy != ReticulumInterfacePolicy::All ||
!hasReadyWifiGateway();
}
bool ReticulumInterfaceSet::wifiSelectedForRuntime() const
{
if (!wifiAllowed())
{
return false;
}
return config_.reticulum_interface_policy ==
ReticulumInterfacePolicy::WifiGatewayOnly ||
hasReadyWifiGateway();
}
bool ReticulumInterfaceSet::hasConfiguredIpInterface() const
{
if (auto_.isConfigured())
@@ -1969,7 +2036,7 @@ bool ReticulumInterfaceSet::hasConfiguredIpInterface() const
void ReticulumInterfaceSet::syncSharedLoRaRxGate()
{
const bool suppress = !loraEnabled();
const bool suppress = !loraSelectedForRuntime();
#if defined(ARDUINO)
if (shared_lora_rx_suppressed_ == suppress &&
app::AppTasks::isRadioReceiveSuppressed() == suppress)
platform/esp/arduino_common/src/chat/infra/store/sd_protocol_peer_repository.cpp
+23 -92
View File
@@ -1,8 +1,7 @@
#include "platform/esp/arduino_common/chat/infra/store/sd_protocol_peer_repository.h"
#include "platform/esp/arduino_common/storage/persistence_bus_gate.h"
#include "platform/esp/arduino_common/storage/scoped_state_lock.h"
#include "platform/esp/arduino_common/storage/sd_card_runtime.h"
#include "platform/esp/common/shared_spi_bus_arbiter.h"
#include <Arduino.h>
@@ -29,47 +28,7 @@ constexpr std::size_t kProtectedContactCapacity = 4096U;
constexpr std::size_t kPeerHotCacheCapacity[] = {16U, 128U, 64U};
constexpr uint32_t kBootCompactionDeltaThreshold = 1024U;
constexpr std::size_t kPendingFlushBudget = 4U;
constexpr uint32_t kPeerStoreBusResource = 5U;
constexpr uint32_t kPeerStoreBusOwnerId = 0x50454552U; // PEER
constexpr const char* kPeerStoreBusOwner = "protocol_peer_v2";
::platform::esp::common::SharedSpiBusAdapter s_peer_store_bus_adapter(
kPeerStoreBusOwner,
kPeerStoreBusOwnerId);
::platform::esp::common::FixedSharedSpiBusPolicyStrategy s_peer_store_bus_policy(
0,
0,
0,
0);
sys::runtime::StorageBusArbiter s_peer_store_bus_arbiter(
s_peer_store_bus_adapter,
s_peer_store_bus_policy);
class ScopedRepositoryLock final
{
public:
explicit ScopedRepositoryLock(SemaphoreHandle_t mutex)
: mutex_(mutex), locked_(mutex_ && xSemaphoreTake(mutex_, portMAX_DELAY) == pdTRUE)
{
}
~ScopedRepositoryLock()
{
if (locked_)
{
xSemaphoreGive(mutex_);
}
}
bool locked() const
{
return locked_;
}
private:
SemaphoreHandle_t mutex_ = nullptr;
bool locked_ = false;
};
using ScopedRepositoryLock = storage_runtime::ScopedRecursiveStateLock;
bool hasText(const char* text)
{
@@ -110,7 +69,7 @@ const MeshPeerNodeFacts* nodeFacts(const MeshPeerRecord& record)
SdProtocolPeerRepository::SdProtocolPeerRepository(IChatStore& chat_store)
: chat_store_(chat_store),
mutex_(xSemaphoreCreateMutex()),
mutex_(xSemaphoreCreateRecursiveMutex()),
node_store_view_(*this),
contact_store_view_(*this)
{
@@ -142,14 +101,7 @@ MeshPeerDirectoryStatus SdProtocolPeerRepository::begin()
return MeshPeerDirectoryStatus::success();
}
storage_runtime::PersistenceBusGate bus_gate(
s_peer_store_bus_arbiter,
sys::runtime::BusAccessPolicy::DurableCommit,
3000U,
kPeerStoreBusResource,
kPeerStoreBusOwnerId + 1U,
kPeerStoreBusOwnerId);
if (!bus_gate.locked() || !ensureLayout())
if (!ensureLayout())
{
return MeshPeerDirectoryStatus::fail(
MeshPeerDirectoryStatusCode::StorageUnavailable);
@@ -524,17 +476,6 @@ bool SdProtocolPeerRepository::appendPeerDelta(
{
return false;
}
storage_runtime::PersistenceBusGate bus_gate(
s_peer_store_bus_arbiter,
sys::runtime::BusAccessPolicy::BackgroundWorkerBounded,
0U,
kPeerStoreBusResource,
kPeerStoreBusOwnerId + 2U,
kPeerStoreBusOwnerId);
if (!bus_gate.locked())
{
return false;
}
char path[96] = {};
buildProtocolPath(protocol, "peers.delta", path, sizeof(path));
if (!journal_.append(path,
@@ -562,17 +503,6 @@ bool SdProtocolPeerRepository::appendContactDelta(
{
return false;
}
storage_runtime::PersistenceBusGate bus_gate(
s_peer_store_bus_arbiter,
sys::runtime::BusAccessPolicy::DurableCommit,
250U,
kPeerStoreBusResource,
kPeerStoreBusOwnerId + 3U,
kPeerStoreBusOwnerId);
if (!bus_gate.locked())
{
return false;
}
char path[96] = {};
buildProtocolPath(protocol, "contacts.delta", path, sizeof(path));
if (!journal_.append(path,
@@ -993,7 +923,12 @@ MeshPeerDirectoryStatus SdProtocolPeerRepository::search(
MeshPeerDirectoryStatusCode::InvalidArgument);
}
protocol = normalizeProtocol(protocol);
*out_count = 0U;
using PeerPtrVector = std::vector<
const MeshPeerRecord*,
::platform::esp::arduino_common::memory::PsramAllocator<
const MeshPeerRecord*>>;
PeerPtrVector matches;
matches.reserve(peers_.size());
for (const MeshPeerRecord& peer : peers_)
{
const MeshPeerNodeFacts* facts = nodeFacts(peer);
@@ -1006,15 +941,18 @@ MeshPeerDirectoryStatus SdProtocolPeerRepository::search(
{
continue;
}
if (*out_count < max_records)
{
out_records[*out_count] = peer;
}
++(*out_count);
if (*out_count >= max_records)
{
break;
}
matches.push_back(&peer);
}
std::sort(matches.begin(),
matches.end(),
[](const MeshPeerRecord* lhs, const MeshPeerRecord* rhs)
{
return lhs->last_seen_s > rhs->last_seen_s;
});
*out_count = std::min(max_records, matches.size());
for (std::size_t index = 0U; index < *out_count; ++index)
{
out_records[index] = *matches[index];
}
return MeshPeerDirectoryStatus::success();
}
@@ -1122,14 +1060,7 @@ MeshPeerDirectoryStatus SdProtocolPeerRepository::clearProtocol(
pending_peer_deltas_.end());
overlayContactFacts();
storage_runtime::PersistenceBusGate bus_gate(
s_peer_store_bus_arbiter,
sys::runtime::BusAccessPolicy::DurableCommit,
1000U,
kPeerStoreBusResource,
kPeerStoreBusOwnerId + 4U,
kPeerStoreBusOwnerId);
if (!bus_gate.locked() || !rewritePeerSnapshot(protocol))
if (!rewritePeerSnapshot(protocol))
{
return MeshPeerDirectoryStatus::fail(MeshPeerDirectoryStatusCode::IoError);
}
platform/esp/arduino_common/src/chat/infra/store/sd_store.cpp
+75 -80
View File
@@ -5,9 +5,8 @@
#include "platform/esp/arduino_common/chat/infra/store/sd_store.h"
#include "platform/esp/arduino_common/storage/persistence_bus_gate.h"
#include "platform/esp/arduino_common/storage/scoped_state_lock.h"
#include "platform/esp/arduino_common/storage/sd_card_runtime.h"
#include "platform/esp/common/shared_spi_bus_arbiter.h"
#if defined(ARDUINO)
#include <Arduino.h>
@@ -30,21 +29,6 @@ namespace storage_v2 = ::chat::storage::v2;
#define CHAT_STORE_LOG(...) std::printf(__VA_ARGS__)
#endif
constexpr uint32_t kChatStoreBusResource = 5;
constexpr uint32_t kChatStoreBusOwnerId = 0x43484154U; // CHAT
constexpr const char* kChatStoreBusOwner = "chat_store_v2";
::platform::esp::common::SharedSpiBusAdapter s_chat_store_bus_adapter(
kChatStoreBusOwner,
kChatStoreBusOwnerId);
::platform::esp::common::FixedSharedSpiBusPolicyStrategy s_chat_store_bus_policy(
0,
0,
0,
0);
sys::runtime::StorageBusArbiter s_chat_store_bus_arbiter(s_chat_store_bus_adapter,
s_chat_store_bus_policy);
constexpr MeshProtocol kProtocols[] = {
MeshProtocol::Meshtastic,
MeshProtocol::MeshCore,
@@ -159,20 +143,14 @@ void hashToHex(const uint8_t* hash, char* out, std::size_t out_len)
} // namespace
SdStore::SdStore()
: mutex_(xSemaphoreCreateRecursiveMutex())
{
scratch_.resize(kScratchCapacity);
catalog_.reserve(64);
read_state_.reserve(64);
statuses_.reserve(256);
seen_hot_.reserve(256);
storage_runtime::PersistenceBusGate startup_gate(
s_chat_store_bus_arbiter,
sys::runtime::BusAccessPolicy::DurableCommit,
3000U,
kChatStoreBusResource,
kChatStoreBusOwnerId + 9U,
kChatStoreBusOwnerId);
ready_ = startup_gate.locked() && ensureLayout() && loadRuntimeState();
ready_ = mutex_ && ensureLayout() && loadRuntimeState();
if (ready_)
{
for (MeshProtocol protocol : kProtocols)
@@ -191,6 +169,15 @@ SdStore::SdStore()
static_cast<unsigned>(seen_hot_.size()));
}
SdStore::~SdStore()
{
if (mutex_)
{
vSemaphoreDelete(mutex_);
mutex_ = nullptr;
}
}
void SdStore::append(const ChatMessage& msg)
{
if (!appendDurably(msg))
@@ -213,14 +200,8 @@ bool SdStore::appendIncomingDurably(const ChatMessage& msg)
bool SdStore::appendInternal(const ChatMessage& input, bool incoming_commit)
{
storage_runtime::PersistenceBusGate transaction_gate(
s_chat_store_bus_arbiter,
sys::runtime::BusAccessPolicy::DurableCommit,
0,
kChatStoreBusResource,
kChatStoreBusOwnerId + 1U,
kChatStoreBusOwnerId);
if (!transaction_gate.locked() || !ready_)
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked() || !ready_)
{
return false;
}
@@ -324,6 +305,15 @@ std::vector<ChatMessage> SdStore::loadPageFromLatest(
std::size_t limit,
std::size_t* total)
{
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked() || !ready_)
{
if (total)
{
*total = 0U;
}
return {};
}
ConversationId conversation = input;
conversation.protocol = normalizeProtocol(conversation.protocol);
const uint32_t count = messageCountOnDisk(conversation);
@@ -357,6 +347,15 @@ std::vector<ConversationMeta> SdStore::loadConversationPage(
std::size_t limit,
std::size_t* total)
{
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked() || !ready_)
{
if (total)
{
*total = 0U;
}
return {};
}
std::vector<ConversationMeta> result;
result.reserve(catalog_.size());
for (const storage_v2::ChatCatalogProjection& projection : catalog_)
@@ -395,6 +394,15 @@ std::vector<ConversationMeta> SdStore::loadConversationPageForProtocol(
std::size_t limit,
std::size_t* total)
{
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked() || !ready_)
{
if (total)
{
*total = 0U;
}
return {};
}
protocol = normalizeProtocol(protocol);
std::vector<ConversationMeta> result;
for (const storage_v2::ChatCatalogProjection& projection : catalog_)
@@ -430,14 +438,8 @@ std::vector<ConversationMeta> SdStore::loadConversationPageForProtocol(
bool SdStore::setUnread(const ConversationId& input, int unread)
{
storage_runtime::PersistenceBusGate transaction_gate(
s_chat_store_bus_arbiter,
sys::runtime::BusAccessPolicy::DurableCommit,
0,
kChatStoreBusResource,
kChatStoreBusOwnerId + 2U,
kChatStoreBusOwnerId);
if (!transaction_gate.locked() || !ready_)
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked() || !ready_)
{
return false;
}
@@ -478,6 +480,11 @@ bool SdStore::setUnread(const ConversationId& input, int unread)
int SdStore::getUnread(const ConversationId& input) const
{
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked() || !ready_)
{
return 0;
}
ConversationId conversation = input;
conversation.protocol = normalizeProtocol(conversation.protocol);
const storage_v2::ChatCatalogProjection* projection =
@@ -487,14 +494,8 @@ int SdStore::getUnread(const ConversationId& input) const
void SdStore::clearConversation(const ConversationId& input)
{
storage_runtime::PersistenceBusGate transaction_gate(
s_chat_store_bus_arbiter,
sys::runtime::BusAccessPolicy::DurableCommit,
0,
kChatStoreBusResource,
kChatStoreBusOwnerId + 3U,
kChatStoreBusOwnerId);
if (!transaction_gate.locked() || !ready_)
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked() || !ready_)
{
return;
}
@@ -544,14 +545,8 @@ void SdStore::clearConversation(const ConversationId& input)
void SdStore::clearAll()
{
storage_runtime::PersistenceBusGate transaction_gate(
s_chat_store_bus_arbiter,
sys::runtime::BusAccessPolicy::RecoveryExclusive,
0,
kChatStoreBusResource,
kChatStoreBusOwnerId + 4U,
kChatStoreBusOwnerId);
if (!transaction_gate.locked())
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked())
{
return;
}
@@ -581,6 +576,11 @@ bool SdStore::updateMessageStatusForProtocol(MessageId msg_id,
MeshProtocol protocol,
MessageStatus status)
{
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked() || !ready_)
{
return false;
}
protocol = normalizeProtocol(protocol);
if (msg_id == 0U || !storage_v2::supportedProtocol(protocol))
{
@@ -593,17 +593,6 @@ bool SdStore::updateMessageStatusForProtocol(MessageId msg_id,
return false;
}
storage_runtime::PersistenceBusGate transaction_gate(
s_chat_store_bus_arbiter,
sys::runtime::BusAccessPolicy::DurableCommit,
0,
kChatStoreBusResource,
kChatStoreBusOwnerId + 5U,
kChatStoreBusOwnerId);
if (!transaction_gate.locked())
{
return false;
}
storage_v2::ChatStatusProjection projection{};
projection.message_id = msg_id;
projection.status = status;
@@ -648,6 +637,11 @@ bool SdStore::updateMessageStatusForProtocol(MessageId msg_id,
bool SdStore::getMessage(MessageId msg_id, ChatMessage* out) const
{
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked() || !ready_)
{
return false;
}
for (MeshProtocol protocol : kProtocols)
{
if (getMessageForProtocol(msg_id, protocol, out))
@@ -662,6 +656,11 @@ bool SdStore::getMessageForProtocol(MessageId msg_id,
MeshProtocol protocol,
ChatMessage* out) const
{
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked() || !ready_)
{
return false;
}
protocol = normalizeProtocol(protocol);
if (msg_id == 0U)
{
@@ -697,6 +696,11 @@ bool SdStore::getMessageForProtocol(MessageId msg_id,
bool SdStore::hasReticulumLxmfMessageHash(const uint8_t* hash) const
{
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked() || !ready_)
{
return false;
}
if (!hash || isAllZeroKeyBytes(hash, kReticulumLxmfHashSize))
{
return false;
@@ -748,7 +752,8 @@ bool SdStore::hasReticulumLxmfMessageHash(const uint8_t* hash) const
void SdStore::flush()
{
if (!ready_)
storage_runtime::ScopedRecursiveStateLock state_lock(mutex_);
if (!state_lock.locked() || !ready_)
{
return;
}
@@ -777,17 +782,7 @@ void SdStore::flush()
return;
}
last_projection_retry_ms_ = now_ms;
storage_runtime::PersistenceBusGate maintenance_gate(
s_chat_store_bus_arbiter,
sys::runtime::BusAccessPolicy::BackgroundWorkerBounded,
0U,
kChatStoreBusResource,
kChatStoreBusOwnerId + 10U,
kChatStoreBusOwnerId);
if (maintenance_gate.locked())
{
(void)compactProtocolProjections(protocol);
}
(void)compactProtocolProjections(protocol);
}
bool SdStore::ensureLayout() const
platform/esp/arduino_common/src/exclusive_lora_runtime.cpp
+5 -1
View File
@@ -33,7 +33,11 @@ bool tryAcquire(Session* out_session)
if (!app::AppTasks::areRadioTasksPaused())
{
app::AppTasks::pauseRadioTasks();
if (!app::AppTasks::pauseRadioTasks())
{
*out_session = {};
return false;
}
out_session->paused_radio_tasks = true;
}
platform/esp/arduino_common/src/platform_ui_reticulum_directory_runtime.cpp
+4 -61
View File
@@ -3,9 +3,7 @@
#if defined(ARDUINO)
#include "chat/ports/i_mesh_peer_directory.h"
#include "platform/esp/arduino_common/storage/persistence_bus_gate.h"
#include "platform/esp/arduino_common/storage/sd_card_runtime.h"
#include "platform/esp/common/shared_spi_bus_arbiter.h"
#include "platform/ui/device_runtime.h"
#include "platform/ui/screen_runtime.h"
#elif defined(ESP_PLATFORM)
@@ -1951,25 +1949,7 @@ constexpr const char* kPagesDir = "/fs/trailmate/reticulum/pages";
constexpr const char* kDefaultPagePath = "/page/index.mu";
constexpr uint32_t kPageCacheReadWaitMs = 60;
constexpr uint32_t kPageCacheWriteWaitMs = 120;
#if defined(ARDUINO)
constexpr uint32_t kPageCacheBusResource = 5;
constexpr uint32_t kPageCacheBusOwnerId = 0x4E504147u; // 'NPAG'
constexpr const char* kPageCacheBusOwner = "reticulum_page_cache";
::platform::esp::common::SharedSpiBusAdapter s_page_cache_bus_adapter(
kPageCacheBusOwner,
kPageCacheBusOwnerId);
::platform::esp::common::FixedSharedSpiBusPolicyStrategy s_page_cache_bus_policy(
kPageCacheReadWaitMs,
kPageCacheReadWaitMs,
kPageCacheWriteWaitMs,
kPageCacheWriteWaitMs);
sys::runtime::StorageBusArbiter s_page_cache_bus_arbiter(
s_page_cache_bus_adapter,
s_page_cache_bus_policy);
#else
SemaphoreHandle_t s_page_storage_mutex = nullptr;
#endif
RequestStartHandler s_request_handler = nullptr;
void* s_request_context = nullptr;
@@ -2025,49 +2005,13 @@ enum class PageCacheBusAccess : uint8_t
Write,
};
#if defined(ARDUINO)
class PageCacheBusGate final
{
public:
explicit PageCacheBusGate(PageCacheBusAccess access)
: gate_(s_page_cache_bus_arbiter,
policyFor(access),
waitMsFor(access),
kPageCacheBusResource,
kPageCacheBusOwnerId + static_cast<uint32_t>(access),
kPageCacheBusOwnerId)
{
}
bool locked() const
{
return gate_.locked();
}
private:
static sys::runtime::BusAccessPolicy policyFor(PageCacheBusAccess access)
{
return access == PageCacheBusAccess::Write
? sys::runtime::BusAccessPolicy::DurableCommit
: sys::runtime::BusAccessPolicy::BackgroundWorkerBounded;
}
static uint32_t waitMsFor(PageCacheBusAccess access)
{
return access == PageCacheBusAccess::Write ? kPageCacheWriteWaitMs
: kPageCacheReadWaitMs;
}
::platform::esp::arduino_common::storage::PersistenceBusGate gate_;
};
#else
bool ensure_page_storage_mutex()
{
if (s_page_storage_mutex)
{
return true;
}
s_page_storage_mutex = xSemaphoreCreateMutex();
s_page_storage_mutex = xSemaphoreCreateRecursiveMutex();
return s_page_storage_mutex != nullptr;
}
@@ -2083,15 +2027,15 @@ class PageCacheBusGate final
const uint32_t wait_ms = access == PageCacheBusAccess::Write
? kPageCacheWriteWaitMs
: kPageCacheReadWaitMs;
locked_ = xSemaphoreTake(s_page_storage_mutex, pdMS_TO_TICKS(wait_ms)) ==
pdTRUE;
locked_ = xSemaphoreTakeRecursive(s_page_storage_mutex,
pdMS_TO_TICKS(wait_ms)) == pdTRUE;
}
~PageCacheBusGate()
{
if (locked_)
{
xSemaphoreGive(s_page_storage_mutex);
xSemaphoreGiveRecursive(s_page_storage_mutex);
}
}
@@ -2106,7 +2050,6 @@ class PageCacheBusGate final
private:
bool locked_ = false;
};
#endif
void copy_text(char* out, std::size_t out_len, const char* text)
{
platform/esp/arduino_common/src/platform_ui_usb_support_runtime.cpp
+9 -3
View File
@@ -411,9 +411,15 @@ void prepare_mass_storage_mode()
if (!app::AppTasks::areRadioTasksPaused())
{
app::AppTasks::pauseRadioTasks();
s_radio_tasks_paused_by_usb = true;
USB_MSC_LOG("radio tasks paused for USB mass storage\n");
if (app::AppTasks::pauseRadioTasks())
{
s_radio_tasks_paused_by_usb = true;
USB_MSC_LOG("radio tasks paused for USB mass storage\n");
}
else
{
USB_MSC_LOG("radio task quiesce failed for USB mass storage\n");
}
}
TaskHandle_t gps_task_handle = gps::gps_get_task_handle();
platform/esp/arduino_common/src/walkie_runtime.cpp
+5 -1
View File
@@ -71,7 +71,11 @@ bool tryAcquire(Session* out_session)
out_session->impl = handles.board;
if (!app::AppTasks::areRadioTasksPaused())
{
app::AppTasks::pauseRadioTasks();
if (!app::AppTasks::pauseRadioTasks())
{
*out_session = {};
return false;
}
out_session->paused_radio_tasks = true;
}
modules/core_chat/tests/test_lxmf_announce_scheduler.cpp → platform/esp/arduino_common/tests/test_lxmf_announce_scheduler.cpp
View File
modules/core_chat/tests/test_lxmf_deferred_discovery_queue.cpp → platform/esp/arduino_common/tests/test_lxmf_deferred_discovery_queue.cpp
View File
modules/core_chat/tests/test_lxmf_runtime_budget.cpp → platform/esp/arduino_common/tests/test_lxmf_runtime_budget.cpp
View File
modules/core_chat/tests/test_protocol_chat_storage_v2_codec.cpp → platform/esp/arduino_common/tests/test_protocol_chat_storage_v2_codec.cpp
View File
modules/core_chat/tests/test_protocol_peer_storage_v2_codec.cpp → platform/esp/arduino_common/tests/test_protocol_peer_storage_v2_codec.cpp
View File
scripts/check_reticulum_runtime_budget_policy.py
+46 -13
View File
@@ -14,6 +14,7 @@ LXMF_H = REPO_ROOT / "platform/esp/arduino_common/include/platform/esp/arduino_c
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"
RTDIR_CPP = REPO_ROOT / "platform/esp/arduino_common/src/platform_ui_reticulum_directory_runtime.cpp"
RTIF_CPP = REPO_ROOT / "platform/esp/arduino_common/src/chat/infra/reticulum/reticulum_interfaces.cpp"
LARGE_LOCAL_PATTERN = re.compile(
r"\buint8_t\s+\w+\s*\[\s*"
@@ -48,6 +49,7 @@ def main() -> int:
rt_cpp = RT_CPP.read_text(encoding="utf-8")
rt_h = RT_H.read_text(encoding="utf-8")
rtdir_cpp = RTDIR_CPP.read_text(encoding="utf-8")
rtif_cpp = RTIF_CPP.read_text(encoding="utf-8")
for signature in (
"bool LxmfAdapter::pollIncomingText",
@@ -58,18 +60,49 @@ def main() -> int:
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)")
maybe_persist_body = method_body(lxmf_cpp, "bool LxmfAdapter::maybePersistPeers")
non_force_branch = maybe_persist_body.split("const bool ok = persistPeers()", 1)[0]
if "if (!force)" not in non_force_branch or "return true;" not in non_force_branch:
violations.append("maybePersistPeers(false) must return before persistPeers()")
announce_body = method_body(lxmf_cpp, "bool LxmfAdapter::handleAnnouncePacket")
peer_record_pos = announce_body.find("recordPeerInDirectory(")
if peer_record_pos >= 0:
gate_pos = announce_body.rfind("allow_persistence", 0, peer_record_pos)
if gate_pos < 0:
violations.append(
"announce peer-directory writes must be gated by allow_persistence"
)
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}")
carrier_checks = {
"bool ReticulumInterfaceSet::sendPacket(": (
"loraSelectedForRuntime()",
"wifiSelectedForRuntime()",
),
"bool ReticulumInterfaceSet::pollIncomingPacket": (
"loraSelectedForRuntime()",
"wifiSelectedForRuntime()",
),
"void ReticulumInterfaceSet::handleRawPacket": (
"loraSelectedForRuntime()",
),
"void ReticulumInterfaceSet::syncSharedLoRaRxGate": (
"loraSelectedForRuntime()",
),
}
for signature, required_calls in carrier_checks.items():
body = method_body(rtif_cpp, signature)
for required_call in required_calls:
if required_call not in body:
violations.append(
f"{signature} must use the shared carrier selection owner"
)
route_body = method_body(lxmf_cpp, "bool LxmfAdapter::routeAndSendPacket")
if "isInterfaceSelected(path->interface_id)" not in route_body:
violations.append(
"Reticulum routing must reject paths bound to an unselected carrier"
)
if "void processSendQueue() override;" not in rt_h:
violations.append("ReticulumAdapter must expose processSendQueue() as the runtime pump hook")
@@ -78,23 +111,23 @@ def main() -> int:
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)
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")
if "record_announce(directory_announce)" in announce_body:
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")
record_announce_body = method_body(rtdir_cpp, "Status record_announce(")
record_address_body = method_body(rtdir_cpp, "Status record_lxmf_address(")
if "queue_announce_async(record)" not in record_announce_body:
violations.append("record_announce() must queue, not synchronously upsert SD")
if "queue_lxmf_address_async(record)" not in record_address_body:
violations.append("record_lxmf_address() must queue, not synchronously upsert SD")
if "record_lxmf_address(" in lxmf_cpp:
violations.append(
"Reticulum runtime must write peer facts through PeerDirectoryService"
)
if "read_byte()" in rtdir_cpp:
violations.append("Reticulum directory TSV reads must use chunked LineReader, not read_byte()")