Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 83 additions & 7 deletions src/mesh/Router.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -329,7 +329,7 @@ ErrorCode Router::sendLocal(meshtastic_MeshPacket *p, RxSource src)
printPacket("Enqueued local", p);
// Preserve the trusted origin explicitly. Queueing used to erase src and make a local
// phone/module packet indistinguishable from remote already-decoded ingress.
handleReceived(p, src);
deliverLocal(p, src);
return ERRNO_SHOULD_RELEASE;
} else if (!iface) {
// We must be sending to remote nodes also, fail if no interface found
Expand All @@ -338,9 +338,10 @@ ErrorCode Router::sendLocal(meshtastic_MeshPacket *p, RxSource src)
return ERRNO_NO_INTERFACES;
} else {
// If we are sending a broadcast, we also treat it as if we just received it ourself
// this allows local apps (and PCs) to see broadcasts sourced locally
// this allows local apps (and PCs) to see broadcasts sourced locally. Only the loopback
// handleReceived is deferred when nested; send(p) below still transmits immediately.
if (isBroadcast(p->to)) {
handleReceived(p, src);
deliverLocal(p, src);
}

// don't override if a channel was requested and no need to set it when PKI is enforced
Expand Down Expand Up @@ -1201,19 +1202,94 @@ NodeNum Router::getNodeNum()
return nodeDB->getNodeNum();
}

bool Router::enqueueDeferredLocal(meshtastic_MeshPacket *p, RxSource src)
{
if (deferredLocalCount >= deferredLocalCapacity)
return false;
uint8_t tail = (deferredLocalHead + deferredLocalCount) % deferredLocalCapacity;
deferredLocalQueue[tail].p = p;
deferredLocalQueue[tail].src = src;
deferredLocalCount++;
return true;
}

bool Router::dequeueDeferredLocal(DeferredLocal &out)
{
if (deferredLocalCount == 0)
return false;
out = deferredLocalQueue[deferredLocalHead];
deferredLocalHead = (deferredLocalHead + 1) % deferredLocalCapacity;
deferredLocalCount--;
return true;
}

void Router::deliverLocal(meshtastic_MeshPacket *p, RxSource src)
{
// Top level: handle synchronously, exactly as before the depth guard existed.
if (handleDepth == 0) {
handleReceived(p, src);
return;
}

// Nested: a module sent this from inside callModules(). Defer a copy so the outermost
// handleReceived() drains it once the current dispatch unwinds, instead of stacking another
// handleReceived() frame on top of the module handler (nRF52 stack overflow on config save).
meshtastic_MeshPacket *copy = packetPool.allocCopy(*p);
if (copy && enqueueDeferredLocal(copy, src))
return;

// Pool exhausted or queue full: drop the deferral. Leak-free and degraded but safe - the
// packet still followed its normal non-loopback path (SHOULD_RELEASE, or the TX path for a
// broadcast). Mirrors sendToPhone()'s degrade-on-exhaustion behavior.
if (copy)
packetPool.release(copy);
LOG_WARN("Deferred local queue full/alloc failed, dropping loopback of 0x%08x", p->id);
#ifdef PIO_UNIT_TESTING
deferredLocalDropped++;
#endif
}

/**
* Handle any packet that is received by an interface on this node.
* Note: some packets may merely being passed through this node and will be forwarded elsewhere.
*/
void Router::handleReceived(meshtastic_MeshPacket *p, RxSource src)
{
handleDepth++;
#ifdef PIO_UNIT_TESTING
if (handleDepth > maxHandleDepthObserved)
maxHandleDepthObserved = handleDepth;
#endif

dispatchReceived(p, src);

// Only the outermost frame drains. Deferred packets were produced by modules sending from
// inside dispatchReceived()'s callModules(); process them here, after the triggering frame has
// unwound, so a second handleReceived() never sits on top of a module handler. handleDepth
// stays >= 1 through the drain, so a drained packet whose own modules send more loopback
// packets enqueues them for this same loop rather than recursing: the stack stays flat and
// processing is breadth-first.
if (handleDepth == 1) {
DeferredLocal d;
while (dequeueDeferredLocal(d)) {
dispatchReceived(d.p, d.src);
packetPool.release(d.p);
}
}

handleDepth--;
}

void Router::dispatchReceived(meshtastic_MeshPacket *p, RxSource src)
{
bool skipHandle = false;

// Store a copy of the encrypted packet for MQTT.
// Local, not a class member: handleReceived re-enters itself when a module
// reply broadcast goes through MeshService::sendToMesh -> Router::sendLocal,
// and a member would be silently overwritten without release on the inner
// call. Each invocation now owns its own copy (issue #9632, #10101, #8729).
// Kept as a local (not a class member) so each dispatch owns its own copy. A shared member was
// historically overwritten without release when a module's reply re-entered this path through
// MeshService::sendToMesh -> Router::sendLocal (issues #9632, #10101, #8729). Nested local
// sends are now deferred rather than synchronously re-entrant (see the drain in
// handleReceived()), so this no longer strictly needs to be a local, but it is kept per-call.
DEBUG_HEAP_BEFORE;
meshtastic_MeshPacket *p_encrypted = packetPool.allocCopy(*p);
DEBUG_HEAP_AFTER("Router::handleReceived", p_encrypted);
Expand Down
54 changes: 54 additions & 0 deletions src/mesh/Router.h
Original file line number Diff line number Diff line change
Expand Up @@ -160,8 +160,62 @@ class Router : protected concurrency::OSThread, protected PacketHistory
*/
void handleReceived(meshtastic_MeshPacket *p, RxSource src = RX_SRC_RADIO);

/**
* The body of handleReceived(): decode, run modules, publish to MQTT. Split out so the
* depth-guarded drain in handleReceived() can process a deferred packet without re-entering
* the drain (and without touching handleDepth) - keeping the stack flat.
*/
void dispatchReceived(meshtastic_MeshPacket *p, RxSource src);

/**
* Route a packet addressed to us (or a local broadcast we loop back) into handleReceived().
* Called synchronously at the top level, but if a module sends this from inside callModules()
* (handleDepth > 0) the packet is copied into the deferred queue instead, so we never stack a
* second handleReceived() on top of a module handler - that nesting is what overflows the
* nRF52 task stack on a config save. Does not consume p; the caller's existing free path is
* unchanged.
*/
void deliverLocal(meshtastic_MeshPacket *p, RxSource src);

/// Depth of handleReceived() frames currently on the stack. >0 means a module is dispatching,
/// so a locally-sent loopback packet must be deferred rather than handled synchronously.
uint8_t handleDepth = 0;

/// A local loopback packet whose handleReceived() was deferred because it was produced from
/// inside callModules(). The queue owns the packet; its RxSource travels with it so the drain
/// dispatches it with the origin the sender intended (RX_SRC_LOCAL stays local).
struct DeferredLocal {
meshtastic_MeshPacket *p;
RxSource src;
};

/// Fixed, small ring buffer of deferred local packets. A config save fans out only a few
/// loopback packets (a self-addressed reply plus a nodeinfo/config broadcast or two), so four
/// slots cover the realistic nesting. On overflow the deferral is dropped (the packet still
/// followed its normal non-loopback path) rather than blocking or growing the heap.
static constexpr uint8_t deferredLocalCapacity = 4;
DeferredLocal deferredLocalQueue[deferredLocalCapacity];
uint8_t deferredLocalHead = 0; // index of the oldest queued entry
uint8_t deferredLocalCount = 0; // entries currently queued

/// Queue a deferred local packet. Returns false (and queues nothing) when full.
bool enqueueDeferredLocal(meshtastic_MeshPacket *p, RxSource src);
/// Pop the oldest deferred local packet into out. Returns false when empty.
bool dequeueDeferredLocal(DeferredLocal &out);

/** Frees the provided packet, and generates a NAK indicating the specifed error while sending */
void abortSendAndNak(meshtastic_Routing_Error err, meshtastic_MeshPacket *p);

#ifdef PIO_UNIT_TESTING
public:
/// High-water mark of handleDepth across this Router's life. The deferral must keep it at 1:
/// a nested local send may never re-enter handleReceived() synchronously.
uint8_t maxHandleDepthObserved = 0;
/// Count of deferrals dropped because the queue was full or a copy could not be allocated.
uint32_t deferredLocalDropped = 0;
/// Number of deferred local packets currently queued.
uint8_t deferredLocalPending() const { return deferredLocalCount; }
#endif
};

enum DecodeState { DECODE_SUCCESS, DECODE_FAILURE, DECODE_OPAQUE, DECODE_FATAL, DECODE_POLICY_REJECT };
Expand Down
173 changes: 173 additions & 0 deletions test/test_mesh_module/test_main.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,96 @@ class ReplyIgnoreModule : public MeshModule
}
};

// Sends, from inside callModules(), a self-addressed reply and a local broadcast - the fan-out an
// AdminModule config save performs. Both go out through service->sendToMesh() while a
// handleReceived() frame is live, so both must be deferred rather than handled re-entrantly.
class NestedFanoutModule : public MeshModule
{
public:
NestedFanoutModule() : MeshModule("nested-fanout", meshtastic_PortNum_PRIVATE_APP) {}
uint32_t handleCalls = 0;

protected:
bool wantPacket(const meshtastic_MeshPacket *p) override { return p->decoded.portnum == ourPortNum; }

ProcessMessage handleReceived(const meshtastic_MeshPacket &mp) override
{
(void)mp;
handleCalls++;

// (a) a reply addressed to ourselves - exercises the isToUs() deferral branch
meshtastic_MeshPacket *reply = router->allocForSending();
reply->to = LOCAL_NODE;
reply->decoded.portnum = meshtastic_PortNum_TEXT_MESSAGE_APP;
reply->decoded.request_id = 0xBEEF;
service->sendToMesh(reply); // default src == RX_SRC_LOCAL

// (b) a local broadcast - exercises the broadcast branch (loopback deferred, TX immediate)
meshtastic_MeshPacket *bcast = router->allocForSending();
bcast->to = NODENUM_BROADCAST;
bcast->decoded.portnum = meshtastic_PortNum_TEXT_MESSAGE_APP;
service->sendToMesh(bcast);

return ProcessMessage::STOP;
}
};

// loopbackOk so it also runs for drained RX_SRC_LOCAL packets; each generation sends the next, so
// the drain must keep processing a chain that grows while it is draining.
class ChainSendModule : public MeshModule
{
public:
ChainSendModule() : MeshModule("chain-send", meshtastic_PortNum_PRIVATE_APP) { loopbackOk = true; }
uint32_t handleCalls = 0;
uint32_t maxGeneration = 0;

protected:
bool wantPacket(const meshtastic_MeshPacket *p) override { return p->decoded.portnum == ourPortNum; }

ProcessMessage handleReceived(const meshtastic_MeshPacket &mp) override
{
handleCalls++;
uint32_t generation = mp.decoded.request_id;
if (generation > maxGeneration)
maxGeneration = generation;

if (generation < 3) {
meshtastic_MeshPacket *next = router->allocForSending();
next->to = LOCAL_NODE;
next->decoded.portnum = ourPortNum;
next->decoded.request_id = generation + 1;
service->sendToMesh(next); // default src == RX_SRC_LOCAL
}
return ProcessMessage::STOP;
}
};

// Sends a burst of self-addressed replies from one dispatch, to overflow the deferred queue.
class BurstSendModule : public MeshModule
{
public:
explicit BurstSendModule(uint32_t count) : MeshModule("burst-send", meshtastic_PortNum_PRIVATE_APP), count(count) {}

protected:
bool wantPacket(const meshtastic_MeshPacket *p) override { return p->decoded.portnum == ourPortNum; }

ProcessMessage handleReceived(const meshtastic_MeshPacket &mp) override
{
(void)mp;
for (uint32_t i = 0; i < count; i++) {
meshtastic_MeshPacket *reply = router->allocForSending();
reply->to = LOCAL_NODE;
reply->decoded.portnum = meshtastic_PortNum_TEXT_MESSAGE_APP;
reply->decoded.request_id = 0x100 + i;
service->sendToMesh(reply); // default src == RX_SRC_LOCAL
}
return ProcessMessage::STOP;
}

private:
uint32_t count;
};

static TestModule *testModule;
static meshtastic_MeshPacket testPacket;
static MockNodeDB *mockNodeDB;
Expand Down Expand Up @@ -204,6 +294,21 @@ static void dispatch(meshtastic_PortNum port)
MeshModule::callModules(request);
}

// Dispatch a want_response==false trigger addressed to us through the real router, so a module's
// handler runs inside a live handleReceived() frame (handleDepth == 1) and any packet it sends is
// deferred. requestId is carried in decoded.request_id for modules that key on a generation.
static void dispatchTrigger(meshtastic_PortNum port, uint32_t requestId = 0)
{
meshtastic_MeshPacket trigger = meshtastic_MeshPacket_init_zero;
trigger.from = LOCAL_NODE;
trigger.to = LOCAL_NODE;
trigger.id = 0x7A190000 + requestId;
trigger.which_payload_variant = meshtastic_MeshPacket_decoded_tag;
trigger.decoded.portnum = port;
trigger.decoded.request_id = requestId;
service->sendToMesh(packetPool.allocCopy(trigger), RX_SRC_USER);
}

} // namespace

void setUp(void)
Expand Down Expand Up @@ -541,6 +646,71 @@ static void test_phoneRequest_replyReachesPhone()
TEST_ASSERT_EQUAL_UINT32(0, mockRoutingModule->ackNaks.size());
}

// A module that fans out a self-addressed reply and a local broadcast from inside callModules()
// must not re-enter handleReceived(): the sends are deferred and drained flat (max depth 1).
static void test_nestedLocalSend_isDeferred_notReentrant()
{
auto *mod = registerDispatchModule(new NestedFanoutModule());

dispatchTrigger(meshtastic_PortNum_PRIVATE_APP);

// The module ran once and was never re-entered; the drain left nothing behind.
TEST_ASSERT_EQUAL_UINT32(1, mod->handleCalls);
TEST_ASSERT_EQUAL_UINT8(1, mockRouter->maxHandleDepthObserved);
TEST_ASSERT_EQUAL_UINT8(0, mockRouter->deferredLocalPending());

// The self-addressed reply reached the phone exactly once (composition with #11185's ccToPhone).
meshtastic_MeshPacket *toPhone = mockService->getForPhone();
TEST_ASSERT_NOT_NULL(toPhone);
TEST_ASSERT_EQUAL_UINT32(LOCAL_NODE, toPhone->to);
TEST_ASSERT_EQUAL_UINT32(0xBEEF, toPhone->decoded.request_id);
mockService->releaseToPool(toPhone);
TEST_ASSERT_NULL(mockService->getForPhone()); // the broadcast did not also go to the phone

// The broadcast still reached the radio TX path exactly once.
TEST_ASSERT_EQUAL_UINT32(1, mockRouter->sentPackets.size());
TEST_ASSERT_TRUE(isBroadcast(mockRouter->sentPackets[0].to));
}

// A drained packet whose own module sends again is picked up by the same drain pass, so a chain of
// local sends is processed breadth-first with the stack held flat (max depth 1).
static void test_deferredChain_drainsBreadthFirst()
{
auto *mod = registerDispatchModule(new ChainSendModule());

dispatchTrigger(meshtastic_PortNum_PRIVATE_APP, 1); // generation 1

// Generations 1 -> 2 -> 3 were all processed in the single outermost drain, never nesting.
TEST_ASSERT_EQUAL_UINT32(3, mod->handleCalls);
TEST_ASSERT_EQUAL_UINT32(3, mod->maxGeneration);
TEST_ASSERT_EQUAL_UINT8(1, mockRouter->maxHandleDepthObserved);
TEST_ASSERT_EQUAL_UINT8(0, mockRouter->deferredLocalPending());
}

// Overflowing the fixed deferred queue drops the excess deferrals gracefully: no crash, no leak,
// depth still capped, and every send's phone cc still happens (return codes unchanged).
static void test_deferredQueueOverflow_dropsGracefully()
{
const uint32_t burst = 6; // deferredLocalCapacity (4) + 2 - keep in sync with Router.h

registerDispatchModule(new BurstSendModule(burst));

dispatchTrigger(meshtastic_PortNum_PRIVATE_APP);

// Two deferrals were dropped (queue full) and nothing re-entered handleReceived().
TEST_ASSERT_EQUAL_UINT32(2, mockRouter->deferredLocalDropped);
TEST_ASSERT_EQUAL_UINT8(1, mockRouter->maxHandleDepthObserved);
TEST_ASSERT_EQUAL_UINT8(0, mockRouter->deferredLocalPending()); // drained clean

// Return codes were unchanged: every reply still cc'd to the phone, dropped deferral or not.
uint32_t delivered = 0;
while (auto *toPhone = mockService->getForPhone()) {
mockService->releaseToPool(toPhone);
delivered++;
}
TEST_ASSERT_EQUAL_UINT32(burst, delivered);
}

void setup()
{
initializeTestEnvironment();
Expand All @@ -567,6 +737,9 @@ void setup()
RUN_TEST(test_dispatch_realNeighborInfoCannotShadowTelemetryOwner);
RUN_TEST(test_localReplyToSelf_isDeliveredToPhone);
RUN_TEST(test_phoneRequest_replyReachesPhone);
RUN_TEST(test_nestedLocalSend_isDeferred_notReentrant);
RUN_TEST(test_deferredChain_drainsBreadthFirst);
RUN_TEST(test_deferredQueueOverflow_dropsGracefully);
exit(UNITY_END());
}

Expand Down
Loading