fix(lxmf): advance submitted opportunistic messages out of queued

This commit is contained in:
liu weikai committed 2026-09-30 14:42:15 +08:00
1 parent 03e9ae21cb
commit 178d4dee22
6 files changed
+179 -1

No files matched your search

BIN
View File
Binary file not shown.
+1
View File
@@ -524,6 +524,7 @@ if(BUILD_TESTING)
"${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
"${TRAIL_MATE_REPO_ROOT}/third_party/rweather_crypto"
"${TRAIL_MATE_REPO_ROOT}/modules/core_chat/include"
"${TRAIL_MATE_REPO_ROOT}/modules/core_sys/include"
"${TRAIL_MATE_REPO_ROOT}/platform/esp/arduino_common/include"
+5
View File
@@ -431,3 +431,8 @@ During synchronization Trail Mate requests the remote transient-ID list,
downloads only missing messages addressed to its local LXMF delivery
destination, and acknowledges handled IDs. Seen transient IDs are retained in
the bounded propagation runtime to suppress duplicate delivery.
For opportunistic direct chat messages, successful transport submission changes
Queued to Sent. A verified recipient proof changes Sent to Delivered. Expiring
proof tracking retains Sent and does not claim delivery; a missing proof is not
evidence that the message remained in the outgoing queue.
@@ -1273,6 +1273,9 @@ bool LxmfAdapter::dispatchLxmfPayload(PeerInfo& peer,
out_dispatch->message_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<unsigned long>(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,
+17
View File
@@ -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)
@@ -0,0 +1,149 @@
#include "platform/esp/arduino_common/chat/infra/lxmf/lxmf_delivery_planner.h"
#include <cassert>
#include <cstddef>
#include <cstdint>
#include <utility>
#include <vector>
// 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<uint8_t>;
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<uint32_t> 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<uint32_t>{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());
}