From 178d4dee22b319bc20144accd28f9bab5bc7b75e Mon Sep 17 00:00:00 2001 From: liu weikai Date: Wed, 30 Sep 2026 14:42:15 +0800 Subject: [PATCH] fix(lxmf): advance submitted opportunistic messages out of queued --- CHANGELOG.md | Bin 123720 -> 124082 bytes apps/linux_sim_shell/CMakeLists.txt | 1 + docs/reticulum_network_config.md | 5 + .../src/chat/infra/lxmf/lxmf_adapter.cpp | 8 +- tests/reticulum_gateway/CMakeLists.txt | 17 ++ .../test_dispatch_status.cpp | 149 ++++++++++++++++++ 6 files changed, 179 insertions(+), 1 deletion(-) create mode 100644 tests/reticulum_gateway/test_dispatch_status.cpp diff --git a/CHANGELOG.md b/CHANGELOG.md index fffbfdd11fa19bc28bb7a33deb3a36587660f7e1..e4b8e0dd16ef1d977224d4ecd7d65e718967172e 100644 GIT binary patch delta 235 zcmX?cjD6Ed_6=t^*nJri8HyOPr`MM;%Cq}0L@@X=xJ_2f6cr9;NM*=lC}B_lDo6v; zsSHKaAJ#L93M()aGn6tUG2}920$IgCGJpC+e@5x)9F2?u$|VdLK&81rZGk{lsbHJ} zmessage_id, millis(), kMaxPendingDeliveryReceipts); + // Transport accepted the packet. A peer proof may later advance + // this to Delivered; waiting for that proof is not queueing. + delivery_notifier_.sent(out_dispatch->message_id); } } @@ -7916,7 +7919,7 @@ void LxmfAdapter::cullTransportState() }); delivery_attempt_ledger_.forEachReceipt( - [now_ms](const runtime::DeliveryAttemptReceipt& receipt) + [this, now_ms](const runtime::DeliveryAttemptReceipt& receipt) { if (receipt.kind != runtime::DeliveryAttemptKind::DirectPacket) { @@ -7937,6 +7940,9 @@ void LxmfAdapter::cullTransportState() destination_hash, static_cast(now_ms - receipt.created_ms)); + // Reconcile the sent state before retiring proof tracking, even + // if the original notification was lost while the bus was full. + delivery_notifier_.sent(receipt.message_id); }); delivery_attempt_ledger_.cull(runtime::DeliveryAttemptKind::DirectPacket, now_ms, diff --git a/tests/reticulum_gateway/CMakeLists.txt b/tests/reticulum_gateway/CMakeLists.txt index 31e050af..d0f2199c 100644 --- a/tests/reticulum_gateway/CMakeLists.txt +++ b/tests/reticulum_gateway/CMakeLists.txt @@ -179,3 +179,20 @@ else() target_compile_options(reticulum_access_config PRIVATE -UNDEBUG -Wall -Wextra) endif() add_test(NAME reticulum_access_config COMMAND reticulum_access_config) + +set(DISPATCH_SOURCE "${REPO}/platform/esp/arduino_common/src/chat/infra/lxmf/lxmf_adapter.cpp") +set_property(DIRECTORY APPEND PROPERTY CMAKE_CONFIGURE_DEPENDS "${DISPATCH_SOURCE}") +file(READ "${DISPATCH_SOURCE}" SOURCE) +extract_between("bool LxmfAdapter::dispatchLxmfPayload(" "const LxmfAdapter::PropagationPeerState*" DISPATCH) +file(WRITE "${CMAKE_CURRENT_BINARY_DIR}/dispatch_actual.inc" "${DISPATCH}") +add_executable(reticulum_dispatch_status test_dispatch_status.cpp + "${REPO}/platform/esp/arduino_common/src/chat/infra/lxmf/lxmf_delivery_planner.cpp") +target_compile_features(reticulum_dispatch_status PRIVATE cxx_std_17) +target_include_directories(reticulum_dispatch_status PRIVATE "${CMAKE_CURRENT_BINARY_DIR}" + "${REPO}/modules/core_chat/include" "${REPO}/platform/esp/arduino_common/include") +if(MSVC) + target_compile_options(reticulum_dispatch_status PRIVATE /UNDEBUG) +else() + target_compile_options(reticulum_dispatch_status PRIVATE -UNDEBUG -Wall -Wextra) +endif() +add_test(NAME reticulum_dispatch_status COMMAND reticulum_dispatch_status) diff --git a/tests/reticulum_gateway/test_dispatch_status.cpp b/tests/reticulum_gateway/test_dispatch_status.cpp new file mode 100644 index 00000000..445d5ea1 --- /dev/null +++ b/tests/reticulum_gateway/test_dispatch_status.cpp @@ -0,0 +1,149 @@ +#include "platform/esp/arduino_common/chat/infra/lxmf/lxmf_delivery_planner.h" +#include +#include +#include +#include +#include + +// Execute the production dispatcher with deterministic transport/crypto seams. +// Cryptographic wire compatibility is covered separately by the vector tests. +namespace reticulum = chat::reticulum; +namespace runtime = chat::lxmf::runtime; +namespace chat::reticulum +{ +inline void computePacketHash(const uint8_t*, size_t, uint8_t*) {} +} // namespace chat::reticulum +namespace chat::lxmf::runtime +{ +using RuntimeByteBuffer = std::vector; +struct DeferredLinkPayload +{ + RuntimeByteBuffer payload; + uint32_t message_id = 0; +}; +} // namespace chat::lxmf::runtime +namespace rtnet +{ +struct Config +{ + struct + { + bool enabled = false; + reticulum::LxmfDeliveryPreference delivery = reticulum::LxmfDeliveryPreference::Automatic; + } propagation; +}; +const Config& active() +{ + static Config config; + return config; +} +} // namespace rtnet +enum class MeshOperationFailure +{ + None, + EncodeFailed, + CryptoFailed, + RadioTxFailed +}; +enum class LocalDestinationKind +{ + Delivery +}; +enum class LinkState +{ + Active, + Pending +}; +struct PeerInfo +{ + uint8_t destination_hash[16]{}; + uint8_t sig_pub[32]{}; +}; +struct LinkSession +{ + LinkState state = LinkState::Pending; +}; +struct OutboundLxmfDispatch +{ + bool ok = false; + bool result_event_deferred = false; + uint32_t message_id = 0; + MeshOperationFailure failure = MeshOperationFailure::None; + uint8_t message_hash[32]{}; + const char* path = "none"; +}; +constexpr size_t kMaxPacketLen = 500; +constexpr size_t kMaxPendingDeliveryReceipts = 8; +uint32_t millis() { return 123; } +uint32_t messageIdFromHash(const uint8_t*) { return 42; } + +struct LxmfAdapter +{ + bool transport_ok = true; + bool ratchet = true; + bool link_available = false; + LinkSession link; + struct Identity + { + const uint8_t* destinationHash() { return nullptr; } + bool sign(const uint8_t*, size_t, uint8_t*) { return true; } + } identity_; + struct Ledger + { + unsigned receipts = 0; + void noteDirectPacketReceipt(const uint8_t*, const uint8_t*, const uint8_t*, uint32_t, uint32_t, size_t) { ++receipts; } + } delivery_attempt_ledger_; + struct Notifier + { + std::vector sent_ids; + void sent(uint32_t id) { sent_ids.push_back(id); } + } delivery_notifier_; + struct Links + { + unsigned queued = 0; + void appendDeferredPayload(LinkSession&, runtime::DeferredLinkPayload&&) { ++queued; } + } link_manager_; + bool shouldRequestPath(PeerInfo&) { return false; } + bool sendPathRequest(PeerInfo&) { return true; } + bool buildSignedPart(const uint8_t*, const uint8_t*, const uint8_t*, size_t, uint8_t*, size_t*, uint8_t*) { return true; } + bool packMessage(const uint8_t*, const uint8_t*, const uint8_t*, const uint8_t*, size_t, uint8_t*, size_t*) { return true; } + LinkSession* findActiveLinkSessionByDestination(const uint8_t*, LocalDestinationKind) { return nullptr; } + bool peerHasUsableRatchet(PeerInfo&) { return ratchet; } + const void* selectActivePropagationPeer() { return nullptr; } + bool queuePropagationUpload(PeerInfo&, const uint8_t*, size_t, uint32_t, const uint8_t*, bool, OutboundLxmfDispatch*) { return false; } + void flushDeferredLinkPayloads(LinkSession&) {} + bool buildSignedMessagePacket(PeerInfo&, const uint8_t*, size_t, uint8_t*, size_t*, uint8_t*) { return true; } + bool routeAndSendPacket(const uint8_t*, size_t, bool) { return transport_ok; } + LinkSession* ensureOutboundLinkSession(PeerInfo&, LocalDestinationKind, bool*) { return link_available ? &link : nullptr; } + bool dispatchLxmfPayload(PeerInfo&, const uint8_t*, size_t, bool, OutboundLxmfDispatch*, bool); +}; +#include "dispatch_actual.inc" + +int main() +{ + PeerInfo peer; + uint8_t payload[]{1, 2, 3}; + OutboundLxmfDispatch result; + LxmfAdapter success; + assert(success.dispatchLxmfPayload(peer, payload, sizeof(payload), true, &result, false)); + assert(success.delivery_attempt_ledger_.receipts == 1); + assert(success.delivery_notifier_.sent_ids == std::vector{42}); + + LxmfAdapter custom; + assert(custom.dispatchLxmfPayload(peer, payload, sizeof(payload), false, &result, false)); + assert(custom.delivery_attempt_ledger_.receipts == 0); + assert(custom.delivery_notifier_.sent_ids.empty()); + + LxmfAdapter failed; + failed.transport_ok = false; + assert(!failed.dispatchLxmfPayload(peer, payload, sizeof(payload), true, &result, false)); + assert(failed.delivery_notifier_.sent_ids.empty()); + + LxmfAdapter deferred; + deferred.transport_ok = false; + deferred.link_available = true; + assert(deferred.dispatchLxmfPayload(peer, payload, sizeof(payload), true, &result, false)); + assert(result.result_event_deferred); + assert(deferred.link_manager_.queued == 1); + assert(deferred.delivery_notifier_.sent_ids.empty()); +}