ElixirForge/firmware
1
0
Fork 0
mirror of https://github.com/alexhopeoconnor/firmware.git synced 2026-10-17 00:51:30 +10:00
Code Releases Activity

Traffic Management module: dedup, rate limiting, role-aware policing (#10706)

Browse Source 
Adds the Traffic Management module (TMM) plus the NodeDB/warm-store and
next-hop foundations it builds on:

- Unified per-node cache (flat array, 8-bit relative ticks) shared by all
  features; role-aware throttles for tracker / lost-and-found.
- Position deduplication: drop unchanged position rebroadcasts within a
  configurable interval; precision driven off the channel ceiling (clamped to
  the public-key max on well-known channels). Enabled by default at 11h.
- Per-node rate limiting and unknown-packet filtering (config-driven; a
  non-zero companion field enables each feature -- no bool toggles).
- NodeInfo direct response from cache with role-based hop clamps.
- Persistent next-hop overflow store: confirmed hops have no TTL, are seeded
  from NodeInfoLite at boot, and survive hot-store eviction.
- Three-tier sender-role resolution (hot NodeInfoLite -> warm store -> TMM
  cache). Role is cached write-time (seeded on first track, refreshed from
  NodeInfo), pins its cache entry like a next-hop hint, and is evicted last.
- Warm store caches device role + protected category across reboot/eviction.
- PositionModule stationary floor for tracker / lost-and-found.
- PSRAM gating for warm/satellite/TMM cache sizes; STM32WL excluded.

Protobufs: TrafficManagementConfig trimmed to the five uint32 fields actually
used; submodule repointed to protobufs develop.

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Tom
2026-06-20 20:35:14 -05:00
committed by GitHub
co-authored by Claude Sonnet 4.6
parent d8d92b7b71
commit c51c01607d
26 changed files with 1186 additions and 610 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/AdminModule.cpp
+4
View File
@@ -481,6 +481,8 @@ bool AdminModule::handleReceivedProtobuf(const meshtastic_MeshPacket &mp, meshta
screen->setFrames(graphics::Screen::FOCUS_PRESERVE); // <-- Rebuild screens
} else if (mp.from == 0) { // local request from the phone — tell the user why it didn't take
sendWarning(NodeDB::PROTECTED_CAP_WARN_FMT, "favorite", r->set_favorite_node, MAX_NUM_NODES - 2);
} else {
LOG_WARN("Remote set_favorite_node for 0x%x refused: protected-node cap", r->set_favorite_node);
}
}
break;
@@ -508,6 +510,8 @@ bool AdminModule::handleReceivedProtobuf(const meshtastic_MeshPacket &mp, meshta
saveChanges(SEGMENT_NODEDATABASE, false);
} else if (mp.from == 0) { // local request from the phone — tell the user why it didn't take
sendWarning(NodeDB::PROTECTED_CAP_WARN_FMT, "ignore", r->set_ignored_node, MAX_NUM_NODES - 2);
} else {
LOG_WARN("Remote set_ignored_node for 0x%x refused: protected-node cap", r->set_ignored_node);
}
}
break;
src/modules/Modules.cpp
+1 -2
View File
@@ -128,8 +128,7 @@ void setupModules()
#endif
#if HAS_TRAFFIC_MANAGEMENT && !MESHTASTIC_EXCLUDE_TRAFFIC_MANAGEMENT
// Instantiate only when enabled to avoid extra memory use and background work.
if (moduleConfig.has_traffic_management && moduleConfig.traffic_management.enabled) {
if (moduleConfig.has_traffic_management) {
trafficManagementModule = new TrafficManagementModule();
}
#endif
src/modules/PositionModule.cpp
+53 -1
View File
@@ -432,6 +432,42 @@ void PositionModule::sendOurPosition(NodeNum dest, bool wantReplies, uint8_t cha
#define RUNONCE_INTERVAL 5000;
bool PositionModule::positionUnchangedSinceLastSend(const meshtastic_PositionLite &selfPos, bool useConfiguredPrecision)
{
if (lastGpsLatitude == 0 && lastGpsLongitude == 0)
return false; // no prior broadcast to compare against
// Broadcast channel = the one sendOurPosition() would pick (first with non-zero on-wire
// precision). Default nodes gauge movement at that on-wire (public-clamped) resolution;
// trackers use their own configured (unclamped) precision so finer moves still count.
uint32_t precisionBits = 0;
for (uint8_t ch = 0; ch < 8; ch++) {
if (getPositionPrecisionForChannel(ch) == 0)
continue;
precisionBits =
useConfiguredPrecision ? getPositionPrecisionForChannel(channels.getByIndex(ch)) : getPositionPrecisionForChannel(ch);
break;
}
return positionWithinPrecisionCell(selfPos.latitude_i, selfPos.longitude_i, lastGpsLatitude, lastGpsLongitude, precisionBits);
}
bool PositionModule::positionWithinPrecisionCell(int32_t aLat, int32_t aLon, int32_t bLat, int32_t bLon, uint32_t precision)
{
if (precision == 0 || precision >= 32)
return false; // sharing disabled or full precision: no coarse cell to hold within
return truncateCoordinate(aLat, precision) == truncateCoordinate(bLat, precision) &&
truncateCoordinate(aLon, precision) == truncateCoordinate(bLon, precision);
}
uint32_t PositionModule::effectiveBroadcastIntervalMs(uint32_t configuredIntervalMs, bool stationary, uint32_t stationaryFloorMs)
{
if (stationary && stationaryFloorMs > configuredIntervalMs)
return stationaryFloorMs;
return configuredIntervalMs;
}
int32_t PositionModule::runOnce()
{
if (sleepOnNextExecution == true) {
@@ -458,7 +494,23 @@ int32_t PositionModule::runOnce()
bool waitingForFreshPosition = (lastGpsSend == 0) && !config.position.fixed_position && !nodeDB->hasLocalPositionSinceBoot();
if (lastGpsSend == 0 || msSinceLastSend >= intervalMs) {
// Hold to the 12h floor when fixed_position (every role: pinning yourself forfeits the
// exception) or when stationary. A real move still goes out early via smart-broadcast below.
// Not-fixed exceptions: lost-and-found broadcasts freely; trackers judge movement at their
// own (unclamped) precision rather than the on-wire one (useConfiguredPrecision).
const auto role = config.device.role;
bool stationary = config.position.fixed_position;
if (!stationary && role != meshtastic_Config_DeviceConfig_Role_LOST_AND_FOUND && nodeDB->hasValidPosition(node)) {
const bool isTracker =
IS_ONE_OF(role, meshtastic_Config_DeviceConfig_Role_TRACKER, meshtastic_Config_DeviceConfig_Role_TAK_TRACKER);
meshtastic_PositionLite selfPos;
if (nodeDB->copyNodePosition(node->num, selfPos))
stationary = positionUnchangedSinceLastSend(selfPos, /*useConfiguredPrecision=*/isTracker);
}
uint32_t effectiveIntervalMs =
effectiveBroadcastIntervalMs(intervalMs, stationary, (uint32_t)default_position_stationary_broadcast_secs * 1000UL);
if (lastGpsSend == 0 || msSinceLastSend >= effectiveIntervalMs) {
if (waitingForFreshPosition) {
#ifdef GPS_DEBUG
LOG_DEBUG("Skip initial position send; no fresh position since boot");
src/modules/PositionModule.h
+14
View File
@@ -38,6 +38,14 @@ class PositionModule : public ProtobufModule<meshtastic_Position>, private concu
void handleNewPosition();
// Pure broadcast-policy helpers, split out so they're unit-testable without the module.
// True when two coordinates truncate to the same precision cell (so a re-broadcast would be a
// duplicate). precision 0 or >=32 returns false: no coarse cell to hold within, never suppress.
static bool positionWithinPrecisionCell(int32_t aLat, int32_t aLon, int32_t bLat, int32_t bLon, uint32_t precision);
// Effective min interval: stationary positions are held to stationaryFloorMs (when that is the
// longer of the two); otherwise the normal configured interval.
static uint32_t effectiveBroadcastIntervalMs(uint32_t configuredIntervalMs, bool stationary, uint32_t stationaryFloorMs);
protected:
/** Called to handle a particular incoming message
@@ -57,6 +65,12 @@ class PositionModule : public ProtobufModule<meshtastic_Position>, private concu
private:
meshtastic_MeshPacket *allocPositionPacket();
struct SmartPosition getDistanceTraveledSinceLastSend(meshtastic_PositionLite currentPosition);
// True when our position is unchanged since the last broadcast: it truncates to the same
// precision grid cell, so re-sending would be a duplicate that traffic management dedups
// downstream anyway. Used to hold stationary broadcasts to a 12h floor. useConfiguredPrecision
// gauges movement at our own configured (unclamped) precision rather than the on-wire
// (public-clamped) precision — trackers report finer movement.
bool positionUnchangedSinceLastSend(const meshtastic_PositionLite &selfPos, bool useConfiguredPrecision);
meshtastic_MeshPacket *allocAtakPli();
void trySetRtc(meshtastic_Position p, bool isLocal, bool forceUpdate = false);
uint32_t precision;
src/modules/TrafficManagementModule.cpp
+293 -257
View File
@@ -2,9 +2,11 @@
#if HAS_TRAFFIC_MANAGEMENT
#include "Channels.h"
#include "Default.h"
#include "MeshService.h"
#include "NodeDB.h"
#include "PositionPrecision.h"
#include "Router.h"
#include "TypeConversions.h"
#include "airtime.h"
@@ -13,6 +15,7 @@
#include "mesh-pb-constants.h"
#include "meshUtils.h"
#include <Arduino.h>
#include <algorithm>
#include <cstring>
#define TM_LOG_DEBUG(fmt, ...) LOG_DEBUG("[TM] " fmt, ##__VA_ARGS__)
@@ -27,7 +30,6 @@ namespace
{
constexpr uint32_t kMaintenanceIntervalMs = 60 * 1000UL; // Cache cleanup interval
constexpr uint32_t kUnknownResetMs = 60 * 1000UL; // Unknown packet window
// NodeInfo direct response: enforced maximum hops by device role
// Both use maxHops logic (respond when hopsAway <= threshold)
@@ -47,6 +49,17 @@ uint32_t secsToMs(uint32_t secs)
return static_cast<uint32_t>(ms);
}
// Advertised role of the originating node (from NodeDB), or CLIENT (no exception) if unknown.
// Position filtering grants two role exceptions: trackers may refresh duplicates hourly, and
// lost-and-found is throttled only to the shortest dedup window. Both are still subject to
// the channel-precision ceiling in alterReceived().
meshtastic_Config_DeviceConfig_Role originRole(NodeNum from)
{
// Resolve via NodeDB: hot store (with user) → warm-tier cached role → CLIENT. The
// warm fallback keeps role exceptions firing for trackers/etc. aged out of the hot store.
return nodeDB ? nodeDB->getNodeRole(from) : meshtastic_Config_DeviceConfig_Role_CLIENT;
}
/**
* Clamp precision to a valid dedup range.
* Invalid values use the module default precision.
@@ -64,57 +77,6 @@ uint8_t sanitizePositionPrecision(uint8_t precision)
return 32;
}
/**
* Check if a timestamp is within a time window.
* Handles wrap-around correctly using unsigned subtraction.
*/
bool isWithinWindow(uint32_t nowMs, uint32_t startMs, uint32_t intervalMs)
{
if (intervalMs == 0 || startMs == 0)
return false;
return (nowMs - startMs) < intervalMs;
}
/**
* Slide an 8-bit relative timestamp back by a wall-clock slab during epoch rebase.
*
* Entries older than the slab clamp to 0 (then reclaimed by the maintenance sweep);
* live entries keep their reconstructed age minus a sub-tick remainder. Each field
* slides by its own resolution's worth of ticks, so a single slab covers all three.
*/
inline void slideRelativeTime(uint8_t &ticks, uint32_t slabMs, uint16_t resolutionSecs)
{
if (ticks == 0 || resolutionSecs == 0)
return;
uint32_t dec = slabMs / (static_cast<uint32_t>(resolutionSecs) * 1000UL);
ticks = (ticks > dec) ? static_cast<uint8_t>(ticks - dec) : 0;
}
/**
* Truncate lat/lon to specified precision for position deduplication.
*
* The truncation works by masking off lower bits and rounding to the center
* of the resulting grid cell. This creates a stable truncated value even
* when GPS jitter causes small coordinate changes.
*
* @param value Raw latitude_i or longitude_i from position
* @param precision Number of significant bits to keep (0-32)
* @return Truncated and centered coordinate value
*/
int32_t truncateLatLon(int32_t value, uint8_t precision)
{
if (precision == 0 || precision >= 32)
return value;
// Create mask to zero out lower bits
uint32_t mask = UINT32_MAX << (32 - precision);
uint32_t truncated = static_cast<uint32_t>(value) & mask;
// Add half the truncation step to center in the grid cell
truncated += (1u << (31 - precision));
return static_cast<int32_t>(truncated);
}
/**
* Saturating increment for uint8_t counters.
* Prevents overflow by capping at UINT8_MAX (255).
@@ -176,23 +138,10 @@ TrafficManagementModule::TrafficManagementModule() : MeshModule("TrafficManageme
encryptedOk = true; // Can process encrypted packets
stats = meshtastic_TrafficManagementStats_init_zero;
// Initialize rolling epoch for relative timestamps
cacheEpochMs = millis();
// Calculate adaptive time resolutions from config (config changes require reboot)
// Resolution = max(60, min(339, interval/2)) for ~24 hour range with good precision
posTimeResolution = calcTimeResolution(Default::getConfiguredOrDefault(
moduleConfig.traffic_management.position_min_interval_secs, default_traffic_mgmt_position_min_interval_secs));
rateTimeResolution = calcTimeResolution(moduleConfig.traffic_management.rate_limit_window_secs);
unknownTimeResolution = calcTimeResolution(kUnknownResetMs / 1000); // ~5 min default
const auto &cfg = moduleConfig.traffic_management;
TM_LOG_INFO("Enabled: pos_dedup=%d nodeinfo_resp=%d rate_limit=%d drop_unknown=%d exhaust_telem=%d exhaust_pos=%d "
"preserve_hops=%d",
cfg.position_dedup_enabled, cfg.nodeinfo_direct_response, cfg.rate_limit_enabled, cfg.drop_unknown_enabled,
cfg.exhaust_hop_telemetry, cfg.exhaust_hop_position, cfg.router_preserve_hops);
TM_LOG_DEBUG("Time resolutions: pos=%us, rate=%us, unknown=%us", posTimeResolution, rateTimeResolution,
unknownTimeResolution);
TM_LOG_INFO("Config: nodeinfo_max_hops=%u rate_window=%us rate_max=%u unknown_thresh=%u pos_interval=%us",
cfg.nodeinfo_direct_response_max_hops, cfg.rate_limit_window_secs, cfg.rate_limit_max_packets,
cfg.unknown_packet_threshold, cfg.position_min_interval_secs);
// Allocate unified cache (10 bytes/entry for all platforms)
#if TRAFFIC_MANAGEMENT_CACHE_SIZE > 0
@@ -278,9 +227,9 @@ void TrafficManagementModule::resetStats()
void TrafficManagementModule::recordRouterHopPreserved()
{
if (!moduleConfig.has_traffic_management || !moduleConfig.traffic_management.enabled)
return;
incrementStat(&stats.router_hops_preserved);
// router_preserve_hops: not suitable right now — removed from config until
// the right heuristic for when to preserve vs. exhaust is clearer.
(void)stats.router_hops_preserved;
}
void TrafficManagementModule::incrementStat(uint32_t *field)
@@ -313,6 +262,18 @@ TrafficManagementModule::UnifiedCacheEntry *TrafficManagementModule::findEntry(N
#endif
}
int TrafficManagementModule::peekCachedRole(NodeNum node)
{
#if TRAFFIC_MANAGEMENT_CACHE_SIZE == 0
(void)node;
return -1;
#else
concurrency::LockGuard guard(&cacheLock);
const UnifiedCacheEntry *entry = findEntry(node);
return entry ? static_cast<int>(entry->getCachedRole()) : -1;
#endif
}
/**
* Find or create an entry for the given node.
*
@@ -326,6 +287,51 @@ TrafficManagementModule::UnifiedCacheEntry *TrafficManagementModule::findEntry(N
* @param isNew Set to true if a new entry was created
* @return Pointer to entry, or nullptr if the cache is unavailable
*/
// Sender-role resolution for the position hot path. The tier-3 cache is authoritative
// here and is kept fresh by updateCachedRoleFromNodeInfo() — i.e. updated at the same
// time NodeDB learns a role, not re-derived on every packet. We only fall back to a
// NodeDB scan (tiers 1+2) the first time we start tracking a node, to seed the cache so
// a resident special-role node is correct from its very first position. Thereafter the
// read is O(1) and survives the node aging out of both NodeDB stores.
meshtastic_Config_DeviceConfig_Role TrafficManagementModule::resolveSenderRole(NodeNum from, UnifiedCacheEntry *entry, bool isNew)
{
if (!entry)
return originRole(from);
if (isNew) {
// First time tracking this node: seed tier 3 from NodeDB (hot → warm). Stores
// CLIENT (0) too, which simply reads back as "no exception".
const meshtastic_Config_DeviceConfig_Role role = originRole(from);
entry->setCachedRole(static_cast<uint8_t>(std::min(15, static_cast<int>(role))));
return role;
}
// Established entry: trust the cached role (refreshed on NodeInfo). No NodeDB scan.
return static_cast<meshtastic_Config_DeviceConfig_Role>(entry->getCachedRole());
}
// Refresh the tier-3 role cache from an observed NodeInfo — the same event that updates
// NodeDB's role — so role changes (including demotion back to CLIENT) are picked up
// without scanning NodeDB on the position hot path. Role is read straight from the
// packet's User payload (authoritative regardless of module ordering). Only updates nodes
// we already track (findEntry, no create) so NodeInfo from non-position nodes can't pollute
// the cache; the role rides along with the node's existing position/rate/unknown state.
void TrafficManagementModule::updateCachedRoleFromNodeInfo(const meshtastic_MeshPacket &mp)
{
#if TRAFFIC_MANAGEMENT_CACHE_SIZE > 0
if (mp.decoded.payload.size == 0)
return;
meshtastic_User user = meshtastic_User_init_zero;
if (!pb_decode_from_bytes(mp.decoded.payload.bytes, mp.decoded.payload.size, &meshtastic_User_msg, &user))
return;
concurrency::LockGuard guard(&cacheLock);
UnifiedCacheEntry *entry = findEntry(getFrom(&mp));
if (entry)
entry->setCachedRole(static_cast<uint8_t>(std::min(15, static_cast<int>(user.role))));
#else
(void)mp;
#endif
}
TrafficManagementModule::UnifiedCacheEntry *TrafficManagementModule::findOrCreateEntry(NodeNum node, bool *isNew)
{
#if TRAFFIC_MANAGEMENT_CACHE_SIZE == 0
@@ -341,7 +347,7 @@ TrafficManagementModule::UnifiedCacheEntry *TrafficManagementModule::findOrCreat
UnifiedCacheEntry *empty = nullptr;
UnifiedCacheEntry *victim = nullptr;
bool victimHasHop = true;
bool leastPreferredVictim = true;
uint8_t victimRecency = UINT8_MAX;
for (uint16_t i = 0; i < cacheSize(); i++) {
@@ -355,15 +361,28 @@ TrafficManagementModule::UnifiedCacheEntry *TrafficManagementModule::findOrCreat
}
if (empty)
continue; // an empty slot beats any victim; stop scoring
const bool hasHop = e.next_hop != 0;
uint8_t recency = e.pos_time;
if (e.rate_time > recency)
recency = e.rate_time;
if (e.unknown_time > recency)
recency = e.unknown_time;
if (!victim || (hasHop == victimHasHop ? recency < victimRecency : !hasHop)) {
// "Preferred" entries are evicted last: a confirmed next-hop hint (routing overflow
// store) or a cached special (non-CLIENT) role (tracker / lost-and-found / router).
// Both are the long-tail state this cache exists to retain.
const bool preferred = e.next_hop != 0 || e.getCachedRole() != meshtastic_Config_DeviceConfig_Role_CLIENT;
// Age in pos-ticks (8-bit modular, wraps correctly). Entries with no
// pos state (pos_time==0) score as maximally old (age=currentPosTick()).
const uint8_t nowPosTick = currentPosTick();
const uint8_t posAge = static_cast<uint8_t>(nowPosTick - e.pos_time);
// Blend in rate/unknown ages scaled to pos-tick units (coarser = conservative).
const uint8_t rateAgePosScale =
static_cast<uint8_t>(static_cast<uint8_t>((currentRateTick() - e.getRateTime()) & 0x0F) * 5 / 3);
const uint8_t unknownAgePosScale =
static_cast<uint8_t>(static_cast<uint8_t>((currentUnknownTick() - e.getUnknownTime()) & 0x0F) / 6);
uint8_t recencyAge = posAge;
if (e.getRateCount() != 0 && rateAgePosScale > recencyAge)
recencyAge = rateAgePosScale;
if (e.getUnknownCount() != 0 && unknownAgePosScale > recencyAge)
recencyAge = unknownAgePosScale;
const uint8_t recency = static_cast<uint8_t>(UINT8_MAX - recencyAge);
if (!victim || (preferred == leastPreferredVictim ? recency < victimRecency : !preferred)) {
victim = &e;
victimHasHop = hasHop;
leastPreferredVictim = preferred;
victimRecency = recency;
}
}
@@ -417,7 +436,7 @@ TrafficManagementModule::NodeInfoPayloadEntry *TrafficManagementModule::findOrCr
NodeInfoPayloadEntry *empty = nullptr;
NodeInfoPayloadEntry *lru = nullptr;
uint32_t lruAge = 0;
const uint32_t now = millis();
const uint32_t now = clockMs();
for (uint16_t i = 0; i < nodeInfoTargetEntries(); i++) {
NodeInfoPayloadEntry &e = nodeInfoPayload[i];
@@ -493,7 +512,7 @@ void TrafficManagementModule::cacheNodeInfoPacket(const meshtastic_MeshPacket &m
// richer context than "just the user protobuf" when PSRAM is present.
// This path is intentionally independent from NodeInfoModule/NodeDB.
entry->user = user;
entry->lastObservedMs = millis();
entry->lastObservedMs = clockMs();
entry->lastObservedRxTime = mp.rx_time;
entry->sourceChannel = mp.channel;
entry->hasDecodedBitfield = mp.decoded.has_bitfield;
@@ -570,11 +589,11 @@ void TrafficManagementModule::clearNextHop(NodeNum dest)
#endif
}
void TrafficManagementModule::preloadNextHopsFromNodeDB()
bool TrafficManagementModule::preloadNextHopsFromNodeDB()
{
#if TRAFFIC_MANAGEMENT_CACHE_SIZE > 0
if (!cache || !nodeDB)
return;
return false; // prerequisites not ready yet — caller should retry on a later pass
uint16_t seeded = 0;
concurrency::LockGuard guard(&cacheLock);
@@ -594,6 +613,9 @@ void TrafficManagementModule::preloadNextHopsFromNodeDB()
}
TM_LOG_INFO("Preloaded %u next-hop hints from NodeDB", static_cast<unsigned>(seeded));
return true;
#else
return true; // nothing to preload on a cache-less build; don't keep retrying
#endif
}
@@ -601,59 +623,11 @@ void TrafficManagementModule::preloadNextHopsFromNodeDB()
// Epoch Management
// =============================================================================
/**
* Reset the timestamp epoch when relative offsets approach overflow.
*
* Called when epoch age exceeds ~19 hours (approaching 8-bit minute overflow).
* Invalidates all cached per-node traffic state.
*/
void TrafficManagementModule::resetEpoch(uint32_t nowMs)
void TrafficManagementModule::flushCache()
{
#if TRAFFIC_MANAGEMENT_CACHE_SIZE > 0
TM_LOG_DEBUG("Resetting cache epoch");
cacheEpochMs = nowMs;
// Full flush avoids stale dedup identity/counters surviving epoch rollover.
TM_LOG_DEBUG("Flushing cache");
memset(cache, 0, static_cast<size_t>(cacheSize()) * sizeof(UnifiedCacheEntry));
#else
(void)nowMs;
#endif
}
/**
* Sliding-epoch rebase — preserve cached state past the 8-bit timestamp horizon.
*
* Instead of flushing the whole cache when offsets approach overflow, advance the
* epoch by a fixed slab and shift every live entry's relative timestamps back by
* the same wall-clock amount. A valid entry's window is only a handful of ticks
* wide (TTL auto-scales with resolution), so live entries comfortably survive;
* already-expired entries clamp to 0 and are reclaimed by the maintenance sweep in
* the same locked pass. Reconstructed absolute time is preserved (minus a sub-tick
* remainder), so in-flight TTL checks remain correct across the rebase.
*
* Caller must hold cacheLock.
*/
void TrafficManagementModule::rebaseEpoch(uint32_t nowMs)
{
#if TRAFFIC_MANAGEMENT_CACHE_SIZE > 0
(void)nowMs;
// Slab stays well below the 200-tick reset threshold so a single rebase drops
// the offset back into range (~200 -> ~72 ticks) while live entries survive.
const uint32_t slabMs = 128UL * maxResolution() * 1000UL;
cacheEpochMs += slabMs;
TM_LOG_DEBUG("Rebasing cache epoch by %lus", static_cast<unsigned long>(slabMs / 1000UL));
for (uint16_t i = 0; i < cacheSize(); i++) {
if (cache[i].node == 0)
continue;
slideRelativeTime(cache[i].pos_time, slabMs, posTimeResolution);
slideRelativeTime(cache[i].rate_time, slabMs, rateTimeResolution);
slideRelativeTime(cache[i].unknown_time, slabMs, unknownTimeResolution);
}
#else
(void)nowMs;
#endif
}
@@ -697,7 +671,12 @@ uint8_t TrafficManagementModule::computePositionFingerprint(int32_t lat_truncate
uint8_t latBits = (static_cast<uint32_t>(lat_truncated) >> shift) & ((1u << bitsToTake) - 1);
uint8_t lonBits = (static_cast<uint32_t>(lon_truncated) >> shift) & ((1u << bitsToTake) - 1);
return static_cast<uint8_t>((latBits << 4) | lonBits);
const uint8_t fp = static_cast<uint8_t>((latBits << 4) | lonBits);
// 0 is the "no position seen" sentinel for pos_fingerprint, so a real position that happens to
// hash to 0 must not collide with it (otherwise its duplicates would never dedup). Remap 0 -> 0xFF,
// mirroring NodeDB::getLastByteOfNodeNum()'s 0 -> 0xFF idiom. Cost: the 0x00 bucket merges into
// 0xFF (one extra collision in 256 — negligible; the fingerprint already collides every 16 cells).
return fp ? fp : 0xFF;
}
// =============================================================================
@@ -712,7 +691,7 @@ uint8_t TrafficManagementModule::computePositionFingerprint(int32_t lat_truncate
// force hop_limit=0 on the rebroadcast copy, allowing one final relay hop.
ProcessMessage TrafficManagementModule::handleReceived(const meshtastic_MeshPacket &mp)
{
if (!moduleConfig.has_traffic_management || !moduleConfig.traffic_management.enabled)
if (!moduleConfig.has_traffic_management)
return ProcessMessage::CONTINUE;
ignoreRequest = false;
@@ -722,7 +701,7 @@ ProcessMessage TrafficManagementModule::handleReceived(const meshtastic_MeshPack
incrementStat(&stats.packets_inspected);
const auto &cfg = moduleConfig.traffic_management;
const uint32_t nowMs = millis();
const uint32_t nowMs = TrafficManagementModule::clockMs();
// -------------------------------------------------------------------------
// Undecoded Packet Handling
@@ -731,7 +710,7 @@ ProcessMessage TrafficManagementModule::handleReceived(const meshtastic_MeshPack
// a misbehaving node. Track and optionally drop repeat offenders.
if (mp.which_payload_variant != meshtastic_MeshPacket_decoded_tag) {
if (cfg.drop_unknown_enabled && cfg.unknown_packet_threshold > 0) {
if (cfg.unknown_packet_threshold > 0) {
if (shouldDropUnknown(&mp, nowMs)) {
logAction("drop", &mp, "unknown");
incrementStat(&stats.unknown_packet_drops);
@@ -742,9 +721,12 @@ ProcessMessage TrafficManagementModule::handleReceived(const meshtastic_MeshPack
return ProcessMessage::CONTINUE;
}
// Learn NodeInfo payloads into the dedicated PSRAM cache.
if (mp.decoded.portnum == meshtastic_PortNum_NODEINFO_APP)
// Learn NodeInfo payloads into the dedicated PSRAM cache, and refresh the tier-3
// role cache for any node we already track (keeps the dedup role exception current).
if (mp.decoded.portnum == meshtastic_PortNum_NODEINFO_APP) {
cacheNodeInfoPacket(mp);
updateCachedRoleFromNodeInfo(mp);
}
// -------------------------------------------------------------------------
// NodeInfo Direct Response
@@ -754,8 +736,8 @@ ProcessMessage TrafficManagementModule::handleReceived(const meshtastic_MeshPack
// STOP prevents the request from being rebroadcast toward the target node,
// and our cached response is sent back to the requestor with hop_limit=0.
if (cfg.nodeinfo_direct_response && mp.decoded.portnum == meshtastic_PortNum_NODEINFO_APP && mp.decoded.want_response &&
!isBroadcast(mp.to) && !isToUs(&mp) && !isFromUs(&mp)) {
if (cfg.nodeinfo_direct_response_max_hops > 0 && mp.decoded.portnum == meshtastic_PortNum_NODEINFO_APP &&
mp.decoded.want_response && !isBroadcast(mp.to) && !isToUs(&mp) && !isFromUs(&mp)) {
if (shouldRespondToNodeInfo(&mp, true)) {
meshtastic_User requester = meshtastic_User_init_zero;
if (pb_decode_from_bytes(mp.decoded.payload.bytes, mp.decoded.payload.size, &meshtastic_User_msg, &requester)) {
@@ -776,7 +758,7 @@ ProcessMessage TrafficManagementModule::handleReceived(const meshtastic_MeshPack
// GPS jitter within the configured precision.
if (!isFromUs(&mp) && !isToUs(&mp)) {
if (cfg.position_dedup_enabled && mp.decoded.portnum == meshtastic_PortNum_POSITION_APP) {
if (channels.isWellKnownChannel(mp.channel) && mp.decoded.portnum == meshtastic_PortNum_POSITION_APP) {
meshtastic_Position pos = meshtastic_Position_init_zero;
if (pb_decode_from_bytes(mp.decoded.payload.bytes, mp.decoded.payload.size, &meshtastic_Position_msg, &pos)) {
if (shouldDropPosition(&mp, &pos, nowMs)) {
@@ -794,7 +776,7 @@ ProcessMessage TrafficManagementModule::handleReceived(const meshtastic_MeshPack
// Throttle nodes sending too many packets within a time window.
// Excludes routing and admin packets which are essential for mesh operation.
if (cfg.rate_limit_enabled && cfg.rate_limit_window_secs > 0 && cfg.rate_limit_max_packets > 0) {
if (cfg.rate_limit_window_secs > 0 && cfg.rate_limit_max_packets > 0) {
if (mp.decoded.portnum != meshtastic_PortNum_ROUTING_APP && mp.decoded.portnum != meshtastic_PortNum_ADMIN_APP) {
if (isRateLimited(mp.from, nowMs)) {
logAction("drop", &mp, "rate-limit");
@@ -811,7 +793,7 @@ ProcessMessage TrafficManagementModule::handleReceived(const meshtastic_MeshPack
void TrafficManagementModule::alterReceived(meshtastic_MeshPacket &mp)
{
if (!moduleConfig.has_traffic_management || !moduleConfig.traffic_management.enabled)
if (!moduleConfig.has_traffic_management)
return;
if (mp.which_payload_variant != meshtastic_MeshPacket_decoded_tag)
@@ -820,40 +802,45 @@ void TrafficManagementModule::alterReceived(meshtastic_MeshPacket &mp)
if (isFromUs(&mp))
return;
// -------------------------------------------------------------------------
// Relayed Broadcast Hop Exhaustion
// -------------------------------------------------------------------------
// For relayed telemetry or position broadcasts from other nodes, optionally
// set hop_limit=0 so they don't propagate further through the mesh.
// exhaust_hop_telemetry / exhaust_hop_position / router_preserve_hops:
// not suitable right now — the right heuristics for when to exhaust or
// preserve hops need more field data before we expose them as config knobs.
// exhaustRequested stays false; perhapsRebroadcast() behaves normally.
const auto &cfg = moduleConfig.traffic_management;
const bool isTelemetry = mp.decoded.portnum == meshtastic_PortNum_TELEMETRY_APP;
const bool isPosition = mp.decoded.portnum == meshtastic_PortNum_POSITION_APP;
// Only exhaust telemetry hops when channel is actually congested, mirroring the same
// airtime checks that gate self-generated telemetry in the telemetry modules.
const bool channelBusy = airTime && (!airTime->isTxAllowedChannelUtil(true) || !airTime->isTxAllowedAirUtil());
const bool shouldExhaust =
((channelBusy && isTelemetry && cfg.exhaust_hop_telemetry) || (isPosition && cfg.exhaust_hop_position));
if (!shouldExhaust || !isBroadcast(mp.to))
return;
if (mp.hop_limit > 0) {
const char *reason = isTelemetry ? "exhaust-hop-telemetry" : "exhaust-hop-position";
logAction("exhaust", &mp, reason);
// Adjust hop_start so downstream nodes compute correct hopsAway (hop_start - hop_limit).
// Without this, hop_limit=0 with original hop_start would show inflated hopsAway.
mp.hop_start = mp.hop_start - mp.hop_limit + 1;
mp.hop_limit = 0;
// Signal perhapsRebroadcast() to allow one final relay with hop_limit=0.
// Without this flag, perhapsRebroadcast() would skip the packet since hop_limit==0.
// The packet-scoped flag is checked in NextHopRouter::perhapsRebroadcast()
// and forces tosend->hop_limit=0, ensuring no further propagation beyond the
// next node.
exhaustRequested = true;
exhaustRequestedFrom = getFrom(&mp);
exhaustRequestedId = mp.id;
incrementStat(&stats.hop_exhausted_packets);
// -------------------------------------------------------------------------
// Relayed Position Precision Clamp
// -------------------------------------------------------------------------
// Clamp relayed position broadcasts to the channel's configured precision
// ceiling. Guards against forwarding more-precise coordinates than the
// channel is intended to carry (e.g. a LongFast channel set to 13-bit /
// ~1.5 km). chanPrec==0 means position sharing is disabled on the channel;
// skip — not our job to zero positions on relay.
// Ham mode (owner.is_licensed) is exempt. Lost-and-found is NOT exempt — its relayed
// positions get the same precision clamp as any node.
// Compile USERPREFS_TMM_APPLY_TO_PRIVATE_CHANNELS to extend to private channels.
if (!owner.is_licensed && isPosition && isBroadcast(mp.to)) {
#ifdef USERPREFS_TMM_APPLY_TO_PRIVATE_CHANNELS
const bool shouldClamp = true;
#else
const bool shouldClamp = channels.isWellKnownChannel(mp.channel);
#endif
if (shouldClamp) {
const uint32_t chanPrec = getPositionPrecisionForChannel(mp.channel);
if (chanPrec > 0) {
meshtastic_Position pos = meshtastic_Position_init_default;
if (pb_decode_from_bytes(mp.decoded.payload.bytes, mp.decoded.payload.size, &meshtastic_Position_msg, &pos)) {
const uint32_t packetPrec = pos.precision_bits > 0 ? pos.precision_bits : 32u;
if (packetPrec > chanPrec) {
applyPositionPrecision(pos, chanPrec);
mp.decoded.payload.size = pb_encode_to_bytes(mp.decoded.payload.bytes, sizeof(mp.decoded.payload.bytes),
&meshtastic_Position_msg, &pos);
logAction("clamp", &mp, "precision");
}
}
}
}
}
}
@@ -863,53 +850,58 @@ void TrafficManagementModule::alterReceived(meshtastic_MeshPacket &mp)
int32_t TrafficManagementModule::runOnce()
{
if (!moduleConfig.has_traffic_management || !moduleConfig.traffic_management.enabled)
if (!moduleConfig.has_traffic_management)
return INT32_MAX;
#if TRAFFIC_MANAGEMENT_CACHE_SIZE > 0
const uint32_t nowMs = millis();
const uint32_t nowMs = TrafficManagementModule::clockMs();
// Warm-start the next-hop cache from persisted NodeInfoLite hints once nodeDB
// is populated. Done here (not in the constructor) so nodeDB has finished
// loading. Takes its own lock, so call before acquiring the sweep guard below.
if (!nextHopPreloaded) {
preloadNextHopsFromNodeDB();
// Only latch the one-shot guard once the preload actually ran; if nodeDB wasn't
// ready yet, retry on the next maintenance pass instead of skipping it forever.
if (!nextHopPreloaded && preloadNextHopsFromNodeDB())
nextHopPreloaded = true;
}
// Calculate TTLs for cache expiration
// Free-running tick counters (no epoch needed).
// TTL expressed in ticks:
// pos: 4× position_min_interval_secs (clamped to 255 ticks @ 6 min/tick)
// rate: 2× rate_limit_window_secs (clamped to 15 ticks @ 5 min/tick; only relevant when rate limits are configured)
// unknown: fixed 12 ticks @ 1 min/tick (only relevant when unknown_packet_threshold > 0)
const uint32_t positionIntervalMs = secsToMs(Default::getConfiguredOrDefault(
moduleConfig.traffic_management.position_min_interval_secs, default_traffic_mgmt_position_min_interval_secs));
const uint32_t positionTtlMs = positionIntervalMs * 4;
const uint8_t posTtlTicks =
static_cast<uint8_t>(std::min(static_cast<uint32_t>(255), (positionIntervalMs * 4) / kPosTimeTickMs));
const uint32_t rateIntervalMs = secsToMs(moduleConfig.traffic_management.rate_limit_window_secs);
const uint32_t rateTtlMs = (rateIntervalMs > 0) ? rateIntervalMs * 2 : (10 * 60 * 1000UL);
const uint32_t rateWindowMs = secsToMs(moduleConfig.traffic_management.rate_limit_window_secs);
const uint8_t rateTtlTicks = static_cast<uint8_t>(
std::min(static_cast<uint32_t>(15), (rateWindowMs > 0 ? rateWindowMs * 2 : 24 * kRateTimeTickMs) / kRateTimeTickMs));
const uint32_t unknownTtlMs = kUnknownResetMs * 5;
// unknown: fixed 12-tick TTL (12 min — 4 ticks past the 5-min default window)
const uint8_t unknownTtlTicks = 12;
const uint8_t nowPosTick = currentPosTick();
const uint8_t nowRateTick = currentRateTick();
const uint8_t nowUnknownTick = currentUnknownTick();
// Sweep cache and clear expired entries
uint16_t activeEntries = 0;
uint16_t expiredEntries = 0;
const uint32_t sweepStartMs = millis();
const uint32_t sweepStartMs = TrafficManagementModule::clockMs();
const auto &cfg = moduleConfig.traffic_management;
concurrency::LockGuard guard(&cacheLock);
// Slide the epoch instead of flushing when offsets approach 8-bit overflow.
// Rebase preserves live entries; only already-expired ones clamp to 0 and are
// reclaimed by the sweep below in this same locked pass.
if (needsEpochReset(nowMs))
rebaseEpoch(nowMs);
for (uint16_t i = 0; i < cacheSize(); i++) {
if (cache[i].node == 0)
continue;
bool anyValid = false;
// Check and clear expired position data
if (cache[i].pos_time != 0) {
uint32_t posTimeMs = fromRelativePosTime(cache[i].pos_time);
if (!isWithinWindow(nowMs, posTimeMs, positionTtlMs)) {
// Check and clear expired position data (presence: pos_fingerprint != 0)
if (cache[i].pos_fingerprint != 0) {
if (static_cast<uint8_t>(nowPosTick - cache[i].pos_time) >= posTtlTicks) {
cache[i].pos_fingerprint = 0;
cache[i].pos_time = 0;
} else {
@@ -917,31 +909,33 @@ int32_t TrafficManagementModule::runOnce()
}
}
// Check and clear expired rate limit data
if (cache[i].rate_time != 0) {
uint32_t rateTimeMs = fromRelativeRateTime(cache[i].rate_time);
if (!isWithinWindow(nowMs, rateTimeMs, rateTtlMs)) {
cache[i].rate_count = 0;
cache[i].rate_time = 0;
// Check and clear expired rate limit data (presence: getRateCount() != 0)
if (cache[i].getRateCount() != 0) {
if ((static_cast<uint8_t>(nowRateTick - cache[i].getRateTime()) & 0x0F) >= rateTtlTicks) {
cache[i].setRateCount(0);
cache[i].setRateTime(0);
} else {
anyValid = true;
}
}
// Check and clear expired unknown tracking data
if (cache[i].unknown_time != 0) {
uint32_t unknownTimeMs = fromRelativeUnknownTime(cache[i].unknown_time);
if (!isWithinWindow(nowMs, unknownTimeMs, unknownTtlMs)) {
cache[i].unknown_count = 0;
cache[i].unknown_time = 0;
// Check and clear expired unknown tracking data (presence: getUnknownCount() != 0)
if (cache[i].getUnknownCount() != 0) {
if ((static_cast<uint8_t>(nowUnknownTick - cache[i].getUnknownTime()) & 0x0F) >= unknownTtlTicks) {
cache[i].setUnknownCount(0);
cache[i].setUnknownTime(0);
} else {
anyValid = true;
}
}
// A confirmed next-hop hint has no TTL of its own and keeps the slot alive,
// so an aged-out routing hint outlives the dedup/rate/unknown state.
if (cache[i].next_hop != 0)
// Two fields have no TTL of their own and pin the slot, so they outlive the
// dedup/rate/unknown state:
// - a confirmed next-hop hint (the routing overflow store), and
// - a cached special (non-CLIENT) role, so a tracker / lost-and-found / router
// keeps its dedup-window exception across quiet periods rather than reverting
// to CLIENT the moment its timed state expires.
if (cache[i].next_hop != 0 || cache[i].getCachedRole() != meshtastic_Config_DeviceConfig_Role_CLIENT)
anyValid = true;
// If all data expired, free the slot entirely
@@ -955,7 +949,7 @@ int32_t TrafficManagementModule::runOnce()
TM_LOG_DEBUG("Maintenance: %u active, %u expired, %u/%u slots, %lums elapsed", activeEntries, expiredEntries,
static_cast<unsigned>(activeEntries), static_cast<unsigned>(cacheSize()),
static_cast<unsigned long>(millis() - sweepStartMs));
static_cast<unsigned long>(TrafficManagementModule::clockMs() - sweepStartMs));
#if defined(ARCH_ESP32) && defined(BOARD_HAS_PSRAM)
if (nodeInfoPayload) {
@@ -984,18 +978,21 @@ bool TrafficManagementModule::shouldDropPosition(const meshtastic_MeshPacket *p,
if (!pos->has_latitude_i || !pos->has_longitude_i)
return false;
uint8_t precision = Default::getConfiguredOrDefault(moduleConfig.traffic_management.position_precision_bits,
default_traffic_mgmt_position_precision_bits);
precision = sanitizePositionPrecision(precision);
// Precision is driven by the channel's own position_precision ceiling — the same
// grid the channel uses for broadcast. Falls back to the firmware default (19-bit,
// ~90m cells) when the channel has no precision configured (chanPrec == 0).
const uint32_t chanPrec = getPositionPrecisionForChannel(p->channel);
uint8_t precision = sanitizePositionPrecision(
chanPrec > 0 ? static_cast<uint8_t>(chanPrec) : static_cast<uint8_t>(default_traffic_mgmt_position_precision_bits));
const int32_t lat_truncated = truncateLatLon(pos->latitude_i, precision);
const int32_t lon_truncated = truncateLatLon(pos->longitude_i, precision);
const int32_t lat_truncated = truncateCoordinate(pos->latitude_i, precision);
const int32_t lon_truncated = truncateCoordinate(pos->longitude_i, precision);
const uint8_t fingerprint = computePositionFingerprint(lat_truncated, lon_truncated, precision);
// Drop gate uses the RAW configured interval: 0 means "dedup disabled" (the
// contract documented below). The 12h default is only for resolution/TTL
// sizing (constructor / runOnce), not for deciding whether to drop — feeding
// the default here would silently turn the 0-disables-dedup contract off.
const uint32_t minIntervalMs = secsToMs(moduleConfig.traffic_management.position_min_interval_secs);
uint32_t minIntervalMs = secsToMs(moduleConfig.traffic_management.position_min_interval_secs);
bool isNew = false;
concurrency::LockGuard guard(&cacheLock);
@@ -1003,19 +1000,46 @@ bool TrafficManagementModule::shouldDropPosition(const meshtastic_MeshPacket *p,
if (!entry)
return false;
// Compare fingerprint and check time window
// When minIntervalMs == 0, deduplication is disabled (withinInterval = false means never drop)
const bool hasPositionState = !isNew && entry->pos_time != 0;
// Role exceptions keyed on the originating node's advertised role, resolved across
// all three tiers (hot store → warm store → TMM live cache). The position path is
// the one place that needs sender-role, and it also keeps tier 3 warm so the
// exception survives the node aging out of both NodeDB stores — important in the
// common dedup-only config, where isRateLimited()'s role write never runs.
const meshtastic_Config_DeviceConfig_Role role = resolveSenderRole(p->from, entry, isNew);
if (role == meshtastic_Config_DeviceConfig_Role_LOST_AND_FOUND) {
// Lost-and-found may refresh a duplicate position at most every ~15 min (cap, never
// lengthens; quantised to ~2 dedup ticks). Only when dedup is active — never tighten
// past an operator who disabled it (0).
const uint32_t lostFoundCapMs = secsToMs(default_traffic_mgmt_lost_and_found_position_min_interval_secs);
if (minIntervalMs != 0 && minIntervalMs > lostFoundCapMs)
minIntervalMs = lostFoundCapMs;
} else if (role == meshtastic_Config_DeviceConfig_Role_TRACKER || role == meshtastic_Config_DeviceConfig_Role_TAK_TRACKER) {
// Trackers may refresh a duplicate position as often as hourly (cap, never lengthens).
const uint32_t trackerCapMs = secsToMs(default_traffic_mgmt_tracker_position_min_interval_secs);
if (minIntervalMs > trackerCapMs)
minIntervalMs = trackerCapMs;
}
// Compare fingerprint and check time window.
// When minIntervalMs == 0, deduplication is disabled (withinInterval = false means never drop).
// Presence: pos_fingerprint != 0; computePositionFingerprint() remaps 0 -> 0xFF so zero means unseen.
const bool hasPositionState = !isNew && entry->pos_fingerprint != 0;
const bool samePosition = hasPositionState && entry->pos_fingerprint == fingerprint;
const uint8_t nowPosTick = currentPosTick();
// Clamp to [1, 255]: intervals shorter than one tick still dedup within the same tick.
const uint8_t windowTicks =
(minIntervalMs == 0) ? 0
: static_cast<uint8_t>(std::min(static_cast<uint32_t>(UINT8_MAX),
std::max(static_cast<uint32_t>(1), minIntervalMs / kPosTimeTickMs)));
const bool withinInterval =
hasPositionState && (minIntervalMs != 0) && isWithinWindow(nowMs, fromRelativePosTime(entry->pos_time), minIntervalMs);
hasPositionState && (windowTicks != 0) && (static_cast<uint8_t>(nowPosTick - entry->pos_time) < windowTicks);
TM_LOG_DEBUG("Position dedup 0x%08x: fp=0x%02x prev=0x%02x same=%d within=%d new=%d", p->from, fingerprint,
entry->pos_fingerprint, samePosition, withinInterval, isNew);
// Update cache entry
// Update cache entry (raw tick; 0 is a valid tick value)
entry->pos_fingerprint = fingerprint;
entry->pos_time = toRelativePosTime(nowMs);
entry->pos_time = nowPosTick;
// Drop only if same position AND within the minimum interval
return samePosition && withinInterval;
@@ -1104,7 +1128,7 @@ bool TrafficManagementModule::shouldRespondToNodeInfo(const meshtastic_MeshPacke
reply->decoded.bitfield |= BITFIELD_OK_TO_MQTT_MASK;
if (hasCachedUser && cachedLastObservedMs != 0) {
uint32_t ageMs = millis() - cachedLastObservedMs;
uint32_t ageMs = clockMs() - cachedLastObservedMs;
TM_LOG_DEBUG("NodeInfo PSRAM hit node=0x%08x age=%lu ms src_ch=%u req_ch=%u rx_time=%lu", p->to,
static_cast<unsigned long>(ageMs), static_cast<unsigned>(cachedSourceChannel),
static_cast<unsigned>(p->channel), static_cast<unsigned long>(cachedLastObservedRxTime));
@@ -1169,25 +1193,32 @@ bool TrafficManagementModule::isRateLimited(NodeNum from, uint32_t nowMs)
if (!entry)
return false;
// Check if window has expired
if (isNew || !isWithinWindow(nowMs, fromRelativeRateTime(entry->rate_time), windowMs)) {
entry->rate_time = toRelativeRateTime(nowMs);
entry->rate_count = 1;
// Window ticks: clamp to [1,15] so zero windowMs (config error) opens a new window.
const uint8_t windowTicks = static_cast<uint8_t>(std::min(static_cast<uint32_t>(15), windowMs / kRateTimeTickMs));
const uint8_t nowRateTick = currentRateTick();
const bool windowExpired =
isNew || entry->getRateCount() == 0 ||
((static_cast<uint8_t>(nowRateTick - entry->getRateTime()) & 0x0F) >= std::max(static_cast<uint8_t>(1), windowTicks));
if (windowExpired) {
entry->setRateTime(nowRateTick);
entry->setRateCount(1);
return false;
}
// Increment counter (saturates at 255)
saturatingIncrement(entry->rate_count);
// Increment counter, saturating at 63 (6-bit field max).
const uint8_t cur = entry->getRateCount();
if (cur < 0x3F)
entry->setRateCount(static_cast<uint8_t>(cur + 1));
// Check against threshold (uint8_t max is 255, but config is uint32_t)
// Threshold capped at 60 so a saturated reading (63) always exceeds it.
uint32_t threshold = moduleConfig.traffic_management.rate_limit_max_packets;
if (threshold > 255)
threshold = 255;
if (threshold > 60)
threshold = 60;
bool limited = entry->rate_count > threshold;
if (limited || entry->rate_count == threshold) {
TM_LOG_DEBUG("Rate limit 0x%08x: count=%u threshold=%u -> %s", from, entry->rate_count, threshold,
limited ? "DROP" : "at-limit");
const uint8_t count = entry->getRateCount();
bool limited = count > threshold;
if (limited || count == threshold) {
TM_LOG_DEBUG("Rate limit 0x%08x: count=%u threshold=%u -> %s", from, count, threshold, limited ? "DROP" : "at-limit");
}
return limited;
#endif
@@ -1200,12 +1231,11 @@ bool TrafficManagementModule::shouldDropUnknown(const meshtastic_MeshPacket *p,
(void)nowMs;
return false;
#else
if (!moduleConfig.traffic_management.drop_unknown_enabled || moduleConfig.traffic_management.unknown_packet_threshold == 0)
if (moduleConfig.traffic_management.unknown_packet_threshold == 0)
return false;
uint32_t windowMs = kUnknownResetMs;
if (moduleConfig.traffic_management.rate_limit_window_secs > 0)
windowMs = secsToMs(moduleConfig.traffic_management.rate_limit_window_secs);
// Fixed 5-tick (5 min) unknown window; capped at 12 ticks (12 min max).
static constexpr uint8_t kUnknownWindowTicks = 5;
bool isNew = false;
concurrency::LockGuard guard(&cacheLock);
@@ -1213,25 +1243,31 @@ bool TrafficManagementModule::shouldDropUnknown(const meshtastic_MeshPacket *p,
if (!entry)
return false;
// Check if window has expired
if (isNew || !isWithinWindow(nowMs, fromRelativeUnknownTime(entry->unknown_time), windowMs)) {
entry->unknown_time = toRelativeUnknownTime(nowMs);
entry->unknown_count = 0;
// Check if window has expired (presence: getUnknownCount() != 0)
const uint8_t nowUnknownTick = currentUnknownTick();
const bool windowExpired = isNew || entry->getUnknownCount() == 0 ||
((static_cast<uint8_t>(nowUnknownTick - entry->getUnknownTime()) & 0x0F) >= kUnknownWindowTicks);
if (windowExpired) {
entry->setUnknownTime(nowUnknownTick);
entry->setUnknownCount(0);
}
// Increment counter (saturates at 255). Same saturation handling as
// isRateLimited: without it, a clamped threshold of 255 can never fire.
const bool alreadySaturated = (entry->unknown_count == UINT8_MAX);
saturatingIncrement(entry->unknown_count);
// Increment counter, saturating at 63 (6-bit field max). With threshold
// capped at 60, a saturated reading always exceeds the limit — no special
// already-saturated edge case needed.
const uint8_t cur = entry->getUnknownCount();
if (cur < 0x3F)
entry->setUnknownCount(static_cast<uint8_t>(cur + 1));
// Check against threshold
// Threshold capped at 60 so a saturated reading (63) always exceeds it.
uint32_t threshold = moduleConfig.traffic_management.unknown_packet_threshold;
if (threshold > 255)
threshold = 255;
if (threshold > 60)
threshold = 60;
bool drop = entry->unknown_count > threshold || (alreadySaturated && threshold == 255);
if (drop || entry->unknown_count == threshold) {
TM_LOG_DEBUG("Unknown packets 0x%08x: count=%u threshold=%u -> %s", p->from, entry->unknown_count, threshold,
const uint8_t count = entry->getUnknownCount();
bool drop = count > threshold;
if (drop || count == threshold) {
TM_LOG_DEBUG("Unknown packets 0x%08x: count=%u threshold=%u -> %s", p->from, count, threshold,
drop ? "DROP" : "at-limit");
}
return drop;
src/modules/TrafficManagementModule.h
+102 -128
View File
@@ -22,9 +22,10 @@
* Memory Optimization:
* Uses one flat unified cache (plain array, linear scan) shared by all
* per-node features instead of separate per-feature caches. Timestamps are
* stored as 8-bit relative offsets from a rolling epoch to further reduce
* memory footprint. LoRa packet rates are low enough that an O(n) scan of
* ~1000 11-byte entries is negligible next to packet processing.
* stored as free-running modular tick counters (pos: 8-bit 360 s/tick;
* rate+unknown: paired 4-bit nibbles in one byte) for a 10-byte entry.
* LoRa packet rates are low enough that an O(n) scan of ~1000 entries is
* negligible next to packet processing.
*/
class TrafficManagementModule : public MeshModule, private concurrency::OSThread
{
@@ -54,7 +55,9 @@ class TrafficManagementModule : public MeshModule, private concurrency::OSThread
// Warm-start the next-hop cache from persisted NodeInfoLite hints so confirmed
// hops survive later hot-store (NodeDB) eviction. Idempotent; runs once after
// nodeDB is populated (lazily on first maintenance pass).
void preloadNextHopsFromNodeDB();
// @return true if it actually ran (prereqs met / nothing to do); false if
// prerequisites (cache, nodeDB) weren't ready yet, so the caller should retry.
bool preloadNextHopsFromNodeDB();
/**
* Check if this packet should have its hops exhausted.
@@ -66,81 +69,90 @@ class TrafficManagementModule : public MeshModule, private concurrency::OSThread
return exhaustRequested && exhaustRequestedFrom == getFrom(&mp) && exhaustRequestedId == mp.id;
}
// Injectable monotonic clock (ms). All TMM time reads go through clockMs() so unit tests can
// advance a virtual timebase instead of sleeping real seconds across the 6 min/360 s tick.
// Mirrors HopScalingModule::s_testNowMs. Writable from tests as TrafficManagementModule::s_testNowMs;
// ignored in production (clockMs() returns millis()).
inline static uint32_t s_testNowMs = 0;
#ifdef PIO_UNIT_TESTING
static uint32_t clockMs() { return s_testNowMs; }
#else
static uint32_t clockMs() { return millis(); }
#endif
protected:
ProcessMessage handleReceived(const meshtastic_MeshPacket &mp) override;
bool wantPacket(const meshtastic_MeshPacket *p) override { return true; }
void alterReceived(meshtastic_MeshPacket &mp) override;
int32_t runOnce() override;
// Protected so test shims can force epoch rollover behavior.
void resetEpoch(uint32_t nowMs);
// Sliding-epoch rebase: advance the epoch and shift live entries back by the
// same wall-clock amount instead of flushing, so cached state survives past the
// ~19h horizon. Caller must hold cacheLock.
void rebaseEpoch(uint32_t nowMs);
// Protected so test shims can flush per-node traffic state.
void flushCache();
// Introspection for tests: the cached device role for a node, or -1 if the node has
// no cache entry (distinguishes "not tracked / evicted" from CLIENT == 0).
int peekCachedRole(NodeNum node);
private:
// =========================================================================
// Unified Cache Entry (11 bytes) - Same for ALL platforms
// Unified Cache Entry (10 bytes) - Same for ALL platforms
// =========================================================================
//
// A single compact structure used across ESP32, NRF52, and all other platforms.
// Memory: 11 bytes × TRAFFIC_MANAGEMENT_CACHE_SIZE entries (default 1000 = 11KB)
//
// Position Fingerprinting:
// Instead of storing full coordinates (8 bytes) or a computed hash,
// we store an 8-bit fingerprint derived deterministically from the
// truncated lat/lon. This extracts the lower 4 significant bits from
// each coordinate: fingerprint = (lat_low4 << 4) | lon_low4
//
// Benefits over hash:
// - Adjacent grid cells have sequential fingerprints (no collision)
// - Two positions only collide if 16+ grid cells apart in BOTH dimensions
// - Deterministic: same input always produces same output
//
// Adaptive Timestamp Resolution:
// All timestamps use 8-bit values with adaptive resolution calculated
// from config at startup. Resolution = max(60, min(339, interval/2)).
// - Min 60 seconds ensures reasonable precision
// - Max 339 seconds allows ~24 hour range (255 * 339 = 86445 sec)
// - interval/2 ensures at least 2 ticks per configured interval
//
// Layout:
// [0-3] node - NodeNum (4 bytes)
// [4] pos_fingerprint - 4 bits lat + 4 bits lon (1 byte)
// [5] rate_count - Packets in current window (1 byte)
// [6] unknown_count - Unknown packets count (1 byte)
// [7] pos_time - Position timestamp (1 byte, adaptive resolution)
// [8] rate_time - Rate window start (1 byte, adaptive resolution)
// [9] unknown_time - Unknown tracking start (1 byte, adaptive resolution)
// [10] next_hop - Last-byte relay to reach `node` (1 byte, 0 = none)
// [0-3] node - NodeNum (4 bytes, 0 = empty slot)
// [4] pos_fingerprint - 4 bits lat + 4 bits lon (0 = no position seen)
// [5] rate_count - [7:6] role[3:2] | [5:0] packets in rate window (0 = no window active)
// [6] unknown_count - [7:6] role[1:0] | [5:0] unknown packets in window (0 = no window active)
// [7] pos_time - Position tick (uint8, free-running 360 s/tick)
// [8] rate_unknown_time - [7:4] rate nibble (300 s/tick) | [3:0] unknown nibble (60 s/tick)
// [9] next_hop - Last-byte relay to reach `node` (0 = none)
//
// next_hop semantics:
// A routing hint: the last byte of the NodeNum to use as next hop to reach
// `node`. Written ONLY from NextHopRouter's ACK-confirmed decision (a
// bidirectionally-verified relay), never inferred one-way from relayed
// traffic. The TMM cache acts as an overflow store for confirmed next-hops
// that have aged out of the hot NodeDB (NodeInfoLite). Unlike the other
// fields it has no TTL of its own — it keeps its slot alive (see runOnce)
// and is refreshed only on the next confirmed exchange.
// The 4-bit device role (bits [7:6] of rate_count paired with [7:6] of unknown_count)
// caches the sender's meshtastic_Config_DeviceConfig_Role as a third fallback after the
// hot store and warm store, for nodes evicted from both. Read/written via
// resolveSenderRole(). Max encodable value is 15.
//
// Presence sentinels (no epoch, no +1 offset needed):
// pos active: pos_fingerprint != 0
// rate active: getRateCount() != 0 (low 6 bits only)
// unknown active: getUnknownCount() != 0 (low 6 bits only)
//
// next_hop: routing hint written only from ACK-confirmed NextHopRouter decisions.
// No TTL — keeps the slot alive across maintenance sweeps.
//
#if _meshtastic_Config_DeviceConfig_Role_MAX > 15
#warning "Device role enum max exceeds 15 — TMM 4-bit role cache (rate_count[7:6]/unknown_count[7:6]) will truncate new values"
#endif
struct __attribute__((packed)) UnifiedCacheEntry {
NodeNum node; // 4 bytes - Node identifier (0 = empty slot)
uint8_t pos_fingerprint; // 1 byte - Lower 4 bits of lat + lon
uint8_t rate_count; // 1 byte - Packet count (saturates at 255)
uint8_t unknown_count; // 1 byte - Unknown packet count (saturates at 255)
uint8_t pos_time; // 1 byte - Position timestamp (adaptive resolution)
uint8_t rate_time; // 1 byte - Rate window start (adaptive resolution)
uint8_t unknown_time; // 1 byte - Unknown tracking start (adaptive resolution)
uint8_t next_hop; // 1 byte - Last-byte relay to reach `node` (0 = none). See note below.
NodeNum node;
uint8_t pos_fingerprint;
uint8_t rate_count; // [7:6] = role[3:2], [5:0] = count (max 63)
uint8_t unknown_count; // [7:6] = role[1:0], [5:0] = count (max 63)
uint8_t pos_time;
uint8_t rate_unknown_time;
uint8_t next_hop;
uint8_t getRateCount() const { return rate_count & 0x3F; }
void setRateCount(uint8_t c) { rate_count = static_cast<uint8_t>((rate_count & 0xC0) | (c & 0x3F)); }
uint8_t getUnknownCount() const { return unknown_count & 0x3F; }
void setUnknownCount(uint8_t c) { unknown_count = static_cast<uint8_t>((unknown_count & 0xC0) | (c & 0x3F)); }
uint8_t getCachedRole() const { return static_cast<uint8_t>(((rate_count >> 6) << 2) | (unknown_count >> 6)); }
void setCachedRole(uint8_t role)
{
rate_count = static_cast<uint8_t>((rate_count & 0x3F) | ((role >> 2) << 6));
unknown_count = static_cast<uint8_t>((unknown_count & 0x3F) | ((role & 0x03) << 6));
}
uint8_t getRateTime() const { return (rate_unknown_time >> 4) & 0x0F; }
uint8_t getUnknownTime() const { return rate_unknown_time & 0x0F; }
void setRateTime(uint8_t t) { rate_unknown_time = static_cast<uint8_t>((rate_unknown_time & 0x0F) | ((t & 0x0F) << 4)); }
void setUnknownTime(uint8_t t) { rate_unknown_time = static_cast<uint8_t>((rate_unknown_time & 0xF0) | (t & 0x0F)); }
};
static_assert(sizeof(UnifiedCacheEntry) == 11, "UnifiedCacheEntry should be 11 bytes");
static_assert(sizeof(UnifiedCacheEntry) == 10, "UnifiedCacheEntry should be 10 bytes");
// =========================================================================
// Flat unified cache
// =========================================================================
//
// Plain array, linear scan (same idiom as WarmNodeStore). A lookup walks at
// most cacheSize() × 11 B — microseconds at LoRa packet rates, not worth a
// most cacheSize() × 10 B — microseconds at LoRa packet rates, not worth a
// hash table. Insertion on a full cache evicts the stalest entry,
// preferring entries without a next_hop hint (those are the long-tail
// routing state this cache exists to keep).
@@ -154,82 +166,29 @@ class TrafficManagementModule : public MeshModule, private concurrency::OSThread
static constexpr uint16_t nodeInfoTargetEntries() { return kNodeInfoCacheEntries; }
// =========================================================================
// Adaptive Timestamp Resolution
// Free-Running Tick Counters
// =========================================================================
//
// All timestamps use 8-bit values with adaptive resolution calculated from
// config at startup. This allows ~24 hour range while maintaining precision.
// Timestamps are stored as free-running modular tick counters derived from
// millis(). No epoch anchor needed: modular subtraction gives correct age
// as long as the true age stays below the counter period.
//
// Resolution formula: max(60, min(339, interval/2))
// - 60 sec minimum ensures reasonable precision
// - 339 sec maximum allows 24 hour range (255 * 339 ≈ 86400 sec)
// - interval/2 ensures at least 2 ticks per configured interval
// pos_time : uint8 (256 ticks × 360 s = 25.6 h period; max window 12 h = 120 ticks)
// rate_time : nibble (16 ticks × 300 s = 80 min period; max window 1 h = 12 ticks)
// unknown_time: nibble (16 ticks × 60 s = 16 min period; max window 12 min = 12 ticks)
//
// Since config changes require reboot, resolution is calculated once.
// Presence sentinels (no +1 offset needed; count fields serve as guards):
// pos active: pos_fingerprint != 0 (0 is reserved sentinel; computePositionFingerprint() remaps computed-0 → 0xFF)
// rate active: getRateCount() != 0 (low 6 bits; high 2 bits are cached role)
// unknown active: getUnknownCount() != 0
//
uint32_t cacheEpochMs = 0;
uint16_t posTimeResolution = 60; // Seconds per tick for position
uint16_t rateTimeResolution = 60; // Seconds per tick for rate limiting
uint16_t unknownTimeResolution = 60; // Seconds per tick for unknown tracking
static constexpr uint32_t kPosTimeTickMs = 360'000UL; // 6 min/tick
static constexpr uint32_t kRateTimeTickMs = 300'000UL; // 5 min/tick
static constexpr uint32_t kUnknownTimeTickMs = 60'000UL; // 1 min/tick
// Calculate resolution from configured interval (called once at startup)
static uint16_t calcTimeResolution(uint32_t intervalSecs)
{
// Resolution = interval/2 to ensure at least 2 ticks per interval
// Clamped to [60, 339] for min precision and max 24h range
uint32_t res = (intervalSecs > 0) ? (intervalSecs / 2) : 60;
if (res < 60)
res = 60;
if (res > 339)
res = 339;
return static_cast<uint16_t>(res);
}
// Convert to/from 8-bit relative timestamps with given resolution.
//
// All stored timestamps carry a uniform +1 "presence" offset: a value of 0 is
// reserved for "no timestamp recorded" (which is also the zero-initialized
// state), and stored values 1..255 encode raw ticks 0..254. This keeps the
// 0-means-empty sentinel consistent with memset/calloc zeroing across every
// sub-store, so the maintenance sweep's `_time != 0` presence checks are
// unambiguous (a timestamp recorded in the first tick after the epoch is no
// longer mistaken for an empty slot). The offset is applied here and removed
// on read, so it cancels out in all window math.
uint8_t toRelativeTime(uint32_t nowMs, uint16_t resolutionSecs) const
{
uint32_t ticks = (nowMs - cacheEpochMs) / (resolutionSecs * 1000UL);
return (ticks >= UINT8_MAX) ? UINT8_MAX : static_cast<uint8_t>(ticks + 1);
}
uint32_t fromRelativeTime(uint8_t ticks, uint16_t resolutionSecs) const
{
return (ticks == 0) ? cacheEpochMs : cacheEpochMs + (static_cast<uint32_t>(ticks - 1) * resolutionSecs * 1000UL);
}
// Convenience wrappers for each timestamp type (the +1 presence offset lives
// in the shared converters above, so these are plain pass-throughs).
uint8_t toRelativePosTime(uint32_t nowMs) const { return toRelativeTime(nowMs, posTimeResolution); }
uint32_t fromRelativePosTime(uint8_t t) const { return fromRelativeTime(t, posTimeResolution); }
uint8_t toRelativeRateTime(uint32_t nowMs) const { return toRelativeTime(nowMs, rateTimeResolution); }
uint32_t fromRelativeRateTime(uint8_t t) const { return fromRelativeTime(t, rateTimeResolution); }
uint8_t toRelativeUnknownTime(uint32_t nowMs) const { return toRelativeTime(nowMs, unknownTimeResolution); }
uint32_t fromRelativeUnknownTime(uint8_t t) const { return fromRelativeTime(t, unknownTimeResolution); }
// Coarsest of the per-feature resolutions (seconds per tick).
uint16_t maxResolution() const
{
uint16_t maxRes = posTimeResolution;
if (rateTimeResolution > maxRes)
maxRes = rateTimeResolution;
if (unknownTimeResolution > maxRes)
maxRes = unknownTimeResolution;
return maxRes;
}
// True when relative offsets approach 8-bit overflow.
// With max resolution of 339 sec, 200 ticks = ~19 hours (safe margin for 24h max).
bool needsEpochReset(uint32_t nowMs) const { return (nowMs - cacheEpochMs) > (200UL * maxResolution() * 1000UL); }
static uint8_t currentPosTick() { return static_cast<uint8_t>(clockMs() / kPosTimeTickMs); }
static uint8_t currentRateTick() { return static_cast<uint8_t>((clockMs() / kRateTimeTickMs) & 0x0F); }
static uint8_t currentUnknownTick() { return static_cast<uint8_t>((clockMs() / kUnknownTimeTickMs) & 0x0F); }
// =========================================================================
// Position Fingerprint
// =========================================================================
@@ -309,6 +268,21 @@ class TrafficManagementModule : public MeshModule, private concurrency::OSThread
// Find existing entry (no creation)
UnifiedCacheEntry *findEntry(NodeNum node);
// Resolve a sender's advertised device role for the position hot path. The tier-3
// cache (this entry's getCachedRole) is authoritative and is kept fresh by
// updateCachedRoleFromNodeInfo() — updated when NodeDB learns a role, not re-derived
// per packet. Only on first tracking (isNew) do we scan NodeDB (hot store → warm
// store, via getNodeRole) to seed the cache, so a resident special-role node is
// correct from its first position; after that the read is O(1) and survives the node
// aging out of both NodeDB stores. Caller must hold cacheLock; entry may be null
// (→ NodeDB scan only).
meshtastic_Config_DeviceConfig_Role resolveSenderRole(NodeNum from, UnifiedCacheEntry *entry, bool isNew);
// Refresh the tier-3 role cache from an observed NodeInfo (the same event that updates
// NodeDB's role). Reads role from the packet's User payload; updates only nodes already
// tracked (no entry creation). Takes cacheLock.
void updateCachedRoleFromNodeInfo(const meshtastic_MeshPacket &mp);
// NodeInfo cache operations (flat PSRAM payload array, linear scan)
const NodeInfoPayloadEntry *findNodeInfoEntry(NodeNum node) const;
NodeInfoPayloadEntry *findOrCreateNodeInfoEntry(NodeNum node, bool *usedEmptySlot);