rich/MeshCore
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden message retry replacement and seed testing channel

Browse Source
This commit is contained in:
mikecarper committed 2026-08-07 23:22:31 -07:00
1 parent 33cea30582
commit 26d6e0f99a
9 files changed
+369 -28
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

examples/companion_radio/MyMesh.cpp
+41 -13
View File
@@ -171,6 +171,8 @@ static bool save_filter(const ContactInfo& c);
#endif
#define PUBLIC_GROUP_PSK "izOH6cXN6mrJ5e26oRXNcg=="
// First 16 bytes of SHA-256("#testing"), encoded as base64.
#define TESTING_GROUP_PSK "zeXoLPUVZH3LVHp5pPBl0Q=="
#ifdef ENABLE_USB_INTERFACE
static const char* terminalContactTypeName(uint8_t type) {
@@ -643,12 +645,15 @@ void MyMesh::expireExpectedAcks() {
}
}
MyMesh::AckTableEntry* MyMesh::findPendingTextMessage(const uint8_t text_fingerprint[MAX_HASH_SIZE]) {
MyMesh::AckTableEntry* MyMesh::findPendingTextMessage(
const uint8_t text_fingerprint[MAX_HASH_SIZE], uint32_t message_timestamp) {
expireExpectedAcks();
for (int i = 0; i < EXPECTED_ACK_TABLE_SIZE; i++) {
AckTableEntry& entry = expected_ack_table[i];
if (entry.ack != 0
&& memcmp(entry.text_fingerprint, text_fingerprint, MAX_HASH_SIZE) == 0) {
&& entry.message_timestamp != message_timestamp
&& memcmp(entry.text_fingerprint, text_fingerprint, MAX_HASH_SIZE) == 0
&& hasActiveRetries(entry.retry_key)) {
return &entry;
}
}
@@ -1265,7 +1270,8 @@ MyMesh::MyMesh(mesh::Radio &radio, mesh::RNG &rng, mesh::RTCClock &rtc, SimpleMe
void MyMesh::begin(bool has_display) {
BaseChatMesh::begin();
if (!_store->loadMainIdentity(self_id)) {
const bool is_new_install = !_store->loadMainIdentity(self_id);
if (is_new_install) {
self_id = radio_new_identity(); // create new random identity
int count = 0;
while (count < 10 && (self_id.pub_key[0] == 0x00 || self_id.pub_key[0] == 0xFF)) { // reserved id hashes
@@ -1347,7 +1353,15 @@ void MyMesh::begin(bool has_display) {
updateGpsTelemetryPolicy();
bootstrapRTCfromContacts();
addChannel("Public", PUBLIC_GROUP_PSK); // pre-configure Andy's public channel
if (is_new_install) {
addChannel("#testing", TESTING_GROUP_PSK);
}
// A saved table takes precedence over compiled defaults on later boots.
_store->loadChannels(this);
if (is_new_install) {
// Make the first-boot defaults survive before a companion app changes them.
_store->saveChannels(this);
}
saved_radio_apply_pending = !applySavedRadioParams();
if (!saved_radio_apply_pending) {
@@ -2085,7 +2099,6 @@ void MyMesh::handleCmdFrame(size_t len) {
mesh::Utils::sha256(text_fingerprint, sizeof(text_fingerprint),
recipient->id.pub_key, PUB_KEY_SIZE,
(const uint8_t*)text, strlen(text));
replacement_entry = findPendingTextMessage(text_fingerprint);
}
int result;
@@ -2107,9 +2120,9 @@ void MyMesh::handleCmdFrame(size_t len) {
// so application retries keep one logical server-side timestamp.
msg_timestamp = getRTCClock()->getCurrentTimeUnique();
}
replacement_entry = findPendingTextMessage(text_fingerprint, msg_timestamp);
result = sendMessage(*recipient, msg_timestamp, attempt, text, expected_ack, est_timeout,
packet_retry_key,
replacement_entry != NULL ? replacement_entry->retry_key : NULL);
packet_retry_key, NULL, text_fingerprint);
if (result != MSG_SEND_FAILED && is_room_post) {
room_message_timestamps.remember(text_fingerprint, app_timestamp,
msg_timestamp);
@@ -2127,13 +2140,14 @@ void MyMesh::handleCmdFrame(size_t len) {
AckTableEntry& entry = replacement_entry != NULL
? *replacement_entry
: expected_ack_table[next_ack_idx];
// Reusing a circular-table slot is only ACK bookkeeping. The evicted
// message still owns its lower-level retry sequence; only a semantic
// same-text replacement (cleared above) or a received ACK may cancel it.
// Reusing this circular slot changes only ACK bookkeeping. Retry
// ownership lives in Mesh and ends only on an ACK, retry completion,
// or a successfully queued semantic replacement.
clearExpectedAck(entry, false);
entry.msg_sent = _ms->getMillis(); // add to circular table
entry.expires_at = futureMillis(est_timeout);
entry.ack = expected_ack;
entry.message_timestamp = msg_timestamp;
entry.contact = recipient;
memcpy(entry.text_fingerprint, text_fingerprint, sizeof(entry.text_fingerprint));
memcpy(entry.retry_key, packet_retry_key, sizeof(entry.retry_key));
@@ -3196,7 +3210,8 @@ ContactInfo* MyMesh::getTerminalRecipient() {
}
void MyMesh::rememberTerminalAck(ContactInfo& recipient, const char* text,
uint32_t expected_ack, uint32_t est_timeout,
uint32_t message_timestamp, uint32_t expected_ack,
uint32_t est_timeout,
const uint8_t packet_retry_key[MAX_HASH_SIZE]) {
if (expected_ack == 0) return;
@@ -3205,6 +3220,7 @@ void MyMesh::rememberTerminalAck(ContactInfo& recipient, const char* text,
entry.msg_sent = _ms->getMillis();
entry.expires_at = futureMillis(est_timeout);
entry.ack = expected_ack;
entry.message_timestamp = message_timestamp;
entry.contact = &recipient;
mesh::Utils::sha256(entry.text_fingerprint, sizeof(entry.text_fingerprint),
recipient.id.pub_key, PUB_KEY_SIZE,
@@ -3259,13 +3275,25 @@ void MyMesh::handleTerminalCommand(char* command) {
} else {
uint32_t expected_ack = 0;
uint32_t est_timeout = 0;
uint32_t message_timestamp = getRTCClock()->getCurrentTimeUnique();
uint8_t text_fingerprint[MAX_HASH_SIZE] = { 0 };
uint8_t packet_retry_key[MAX_HASH_SIZE] = { 0 };
int result = sendMessage(*recipient, getRTCClock()->getCurrentTimeUnique(), 0,
text, expected_ack, est_timeout, packet_retry_key);
mesh::Utils::sha256(text_fingerprint, sizeof(text_fingerprint),
recipient->id.pub_key, PUB_KEY_SIZE,
(const uint8_t*)text, strlen(text));
AckTableEntry* replacement_entry =
findPendingTextMessage(text_fingerprint, message_timestamp);
int result = sendMessage(*recipient, message_timestamp, 0, text,
expected_ack, est_timeout, packet_retry_key,
NULL, text_fingerprint);
if (result == MSG_SEND_FAILED) {
Serial.print(" ERROR: unable to send\r\n");
} else {
rememberTerminalAck(*recipient, text, expected_ack, est_timeout, packet_retry_key);
if (replacement_entry != NULL) {
clearExpectedAck(*replacement_entry, false);
}
rememberTerminalAck(*recipient, text, message_timestamp, expected_ack,
est_timeout, packet_retry_key);
Serial.printf(" message sent - %s\r\n",
result == MSG_SEND_SENT_FLOOD ? "FLOOD" : "DIRECT");
}
examples/companion_radio/MyMesh.h
+5 -2
View File
@@ -264,7 +264,8 @@ private:
ContactInfo* getTerminalRecipient();
void importTerminalCard(char* command);
void rememberTerminalAck(ContactInfo& recipient, const char* text,
uint32_t expected_ack, uint32_t est_timeout,
uint32_t message_timestamp, uint32_t expected_ack,
uint32_t est_timeout,
const uint8_t packet_retry_key[MAX_HASH_SIZE]);
#endif
bool isValidClientRepeatFreq(uint32_t f) const;
@@ -349,6 +350,7 @@ private:
unsigned long msg_sent;
unsigned long expires_at;
uint32_t ack;
uint32_t message_timestamp;
ContactInfo* contact;
uint8_t text_fingerprint[MAX_HASH_SIZE];
uint8_t retry_key[MAX_HASH_SIZE];
@@ -365,7 +367,8 @@ private:
void clearExpectedAck(AckTableEntry& entry, bool cancel_retries = true);
void expireExpectedAcks();
AckTableEntry* findPendingTextMessage(const uint8_t text_fingerprint[MAX_HASH_SIZE]);
AckTableEntry* findPendingTextMessage(
const uint8_t text_fingerprint[MAX_HASH_SIZE], uint32_t message_timestamp);
#define ADVERT_PATH_TABLE_SIZE 16
AdvertPath advert_paths[ADVERT_PATH_TABLE_SIZE]; // circular table
examples/simple_room_server/MyMesh.cpp
+30 -10
View File
@@ -107,24 +107,43 @@ void MyMesh::pushPostToClient(ClientInfo *client, PostInfo &post) {
memcpy(&reply_data[len], post.text, text_len);
len += text_len;
uint8_t message_participants[2 * PUB_KEY_SIZE];
memcpy(message_participants, client->id.pub_key, PUB_KEY_SIZE);
memcpy(&message_participants[PUB_KEY_SIZE], post.author.pub_key, PUB_KEY_SIZE);
uint8_t message_retry_key[MAX_HASH_SIZE];
mesh::Utils::sha256(message_retry_key, sizeof(message_retry_key),
message_participants, sizeof(message_participants),
(const uint8_t*)post.text, text_len);
// calc expected ACK reply
mesh::Utils::sha256((uint8_t *)&client->extra.room.pending_ack, 4, reply_data, len, client->id.pub_key, PUB_KEY_SIZE);
client->extra.room.push_post_timestamp = post.post_timestamp;
auto reply = createDatagram(PAYLOAD_TYPE_TXT_MSG, client->id, client->shared_secret, reply_data, len);
bool sent = false;
if (reply) {
if (client->out_path_len == OUT_PATH_UNKNOWN) {
unsigned long delay_millis = 0;
sendFloodScoped(default_scope, reply, delay_millis, _prefs.path_hash_mode + 1); // REVISIT
client->extra.room.ack_timeout = futureMillis(PUSH_ACK_TIMEOUT_FLOOD);
sent = sendFloodScoped(default_scope, reply, delay_millis,
_prefs.path_hash_mode + 1); // REVISIT
if (sent) {
client->extra.room.ack_timeout = futureMillis(PUSH_ACK_TIMEOUT_FLOOD);
}
} else {
sendDirect(reply, client->out_path, client->out_path_len);
sent = sendDirect(reply, client->out_path, client->out_path_len);
uint8_t path_hash_count = client->out_path_len & 63;
client->extra.room.ack_timeout = futureMillis(PUSH_TIMEOUT_BASE + PUSH_ACK_TIMEOUT_FACTOR * (path_hash_count + 1));
if (sent) {
uint8_t path_hash_count = client->out_path_len & 63;
client->extra.room.ack_timeout = futureMillis(
PUSH_TIMEOUT_BASE + PUSH_ACK_TIMEOUT_FACTOR * (path_hash_count + 1));
}
}
_num_post_pushes++; // stats
} else {
if (sent) {
replaceActiveMessageRetries(reply, message_retry_key, post.post_timestamp);
_num_post_pushes++; // stats
}
}
if (!sent) {
client->extra.room.pending_ack = 0;
MESH_DEBUG_PRINTLN("Unable to push post to client");
}
@@ -1308,14 +1327,15 @@ bool MyMesh::applySavedRadioParams() {
return true;
}
void MyMesh::sendFloodScoped(const TransportKey& scope, mesh::Packet* pkt, uint32_t delay_millis, uint8_t path_hash_size) {
bool MyMesh::sendFloodScoped(const TransportKey& scope, mesh::Packet* pkt,
uint32_t delay_millis, uint8_t path_hash_size) {
if (scope.isNull()) {
sendFlood(pkt, delay_millis, path_hash_size);
return sendFlood(pkt, delay_millis, path_hash_size);
} else {
uint16_t codes[2];
codes[0] = scope.calcTransportCode(pkt);
codes[1] = 0; // REVISIT: set to 'home' Region, for sender/return region?
sendFlood(pkt, codes, delay_millis, path_hash_size);
return sendFlood(pkt, codes, delay_millis, path_hash_size);
}
}
examples/simple_room_server/MyMesh.h
+2 -1
View File
@@ -331,7 +331,8 @@ public:
_cli.savePrefs(_fs, scope);
}
void sendFloodScoped(const TransportKey& scope, mesh::Packet* pkt, uint32_t delay_millis, uint8_t path_hash_size);
bool sendFloodScoped(const TransportKey& scope, mesh::Packet* pkt,
uint32_t delay_millis, uint8_t path_hash_size);
// CommonCLICallbacks
void applyTempRadioParams(float freq, float bw, uint8_t sf, uint8_t cr, int timeout_mins) override;
src/Mesh.cpp
+123
View File
@@ -166,9 +166,12 @@ void Mesh::begin() {
_direct_retries[i].echo_wait_started_at = 0;
_direct_retries[i].retry_at = 0;
_direct_retries[i].retry_delay = 0;
_direct_retries[i].message_timestamp = 0;
_direct_retries[i].retry_attempts_sent = 0;
memset(_direct_retries[i].retry_key, 0, sizeof(_direct_retries[i].retry_key));
memset(_direct_retries[i].trace_replacement_key, 0, sizeof(_direct_retries[i].trace_replacement_key));
memset(_direct_retries[i].message_replacement_key, 0,
sizeof(_direct_retries[i].message_replacement_key));
memset(_direct_retries[i].next_hop_hash, 0, sizeof(_direct_retries[i].next_hop_hash));
_direct_retries[i].next_hop_hash_len = 0;
_direct_retries[i].payload_type = 0;
@@ -178,6 +181,7 @@ void Mesh::begin() {
_direct_retries[i].final_hop_retry = false;
_direct_retries[i].waiting_final_echo = false;
_direct_retries[i].queued = false;
_direct_retries[i].has_message_replacement_key = false;
_direct_retries[i].active = false;
}
for (int i = 0; i < MAX_FLOOD_RETRY_SLOTS; i++) {
@@ -186,13 +190,17 @@ void Mesh::begin() {
_flood_retries[i].retry_started_at = 0;
_flood_retries[i].retry_at = 0;
_flood_retries[i].retry_delay = 0;
_flood_retries[i].message_timestamp = 0;
_flood_retries[i].retry_attempts_sent = 0;
memset(_flood_retries[i].retry_key, 0, sizeof(_flood_retries[i].retry_key));
memset(_flood_retries[i].message_replacement_key, 0,
sizeof(_flood_retries[i].message_replacement_key));
_flood_retries[i].priority = 0;
_flood_retries[i].progress_marker = 0;
_flood_retries[i].self_advert = false;
_flood_retries[i].waiting_final_echo = false;
_flood_retries[i].queued = false;
_flood_retries[i].has_message_replacement_key = false;
_flood_retries[i].active = false;
}
for (int i = 0; i < MAX_RECENT_ADVERT_ECHOS; i++) {
@@ -1037,9 +1045,12 @@ void Mesh::clearDirectRetrySlot(int idx) {
_direct_retries[idx].echo_wait_started_at = 0;
_direct_retries[idx].retry_at = 0;
_direct_retries[idx].retry_delay = 0;
_direct_retries[idx].message_timestamp = 0;
_direct_retries[idx].retry_attempts_sent = 0;
memset(_direct_retries[idx].retry_key, 0, sizeof(_direct_retries[idx].retry_key));
memset(_direct_retries[idx].trace_replacement_key, 0, sizeof(_direct_retries[idx].trace_replacement_key));
memset(_direct_retries[idx].message_replacement_key, 0,
sizeof(_direct_retries[idx].message_replacement_key));
memset(_direct_retries[idx].next_hop_hash, 0, sizeof(_direct_retries[idx].next_hop_hash));
_direct_retries[idx].next_hop_hash_len = 0;
_direct_retries[idx].payload_type = 0;
@@ -1049,6 +1060,7 @@ void Mesh::clearDirectRetrySlot(int idx) {
_direct_retries[idx].final_hop_retry = false;
_direct_retries[idx].waiting_final_echo = false;
_direct_retries[idx].queued = false;
_direct_retries[idx].has_message_replacement_key = false;
_direct_retries[idx].active = false;
if (rebuild_timeout) rebuildNextDirectRetryTimeout();
}
@@ -1634,13 +1646,17 @@ void Mesh::clearFloodRetrySlot(int idx) {
_flood_retries[idx].retry_started_at = 0;
_flood_retries[idx].retry_at = 0;
_flood_retries[idx].retry_delay = 0;
_flood_retries[idx].message_timestamp = 0;
_flood_retries[idx].retry_attempts_sent = 0;
memset(_flood_retries[idx].retry_key, 0, sizeof(_flood_retries[idx].retry_key));
memset(_flood_retries[idx].message_replacement_key, 0,
sizeof(_flood_retries[idx].message_replacement_key));
_flood_retries[idx].priority = 0;
_flood_retries[idx].progress_marker = 0;
_flood_retries[idx].self_advert = false;
_flood_retries[idx].waiting_final_echo = false;
_flood_retries[idx].queued = false;
_flood_retries[idx].has_message_replacement_key = false;
_flood_retries[idx].active = false;
if (rebuild_timeout) rebuildNextFloodRetryTimeout();
}
@@ -1794,6 +1810,113 @@ void Mesh::replaceActiveRetries(const Packet* replacement_packet,
}
}
void Mesh::replaceActiveMessageRetries(
const Packet* replacement_packet,
const uint8_t message_key[MAX_HASH_SIZE], uint32_t message_timestamp) {
if (replacement_packet == NULL || message_key == NULL
|| replacement_packet->getPayloadType() != PAYLOAD_TYPE_TXT_MSG) {
return;
}
int replacement_direct_slot = -1;
int replacement_flood_slot = -1;
bool found_prior = false;
for (int i = 0; i < MAX_DIRECT_RETRY_SLOTS; i++) {
if (!_direct_retries[i].active) continue;
if (_direct_retries[i].trigger_packet == replacement_packet) {
replacement_direct_slot = i;
continue;
}
if (_direct_retries[i].has_message_replacement_key
&& _direct_retries[i].message_timestamp != message_timestamp
&& memcmp(_direct_retries[i].message_replacement_key, message_key,
MAX_HASH_SIZE) == 0) {
found_prior = true;
}
}
for (int i = 0; i < MAX_FLOOD_RETRY_SLOTS; i++) {
if (!_flood_retries[i].active) continue;
if (_flood_retries[i].trigger_packet == replacement_packet) {
replacement_flood_slot = i;
continue;
}
if (_flood_retries[i].has_message_replacement_key
&& _flood_retries[i].message_timestamp != message_timestamp
&& memcmp(_flood_retries[i].message_replacement_key, message_key,
MAX_HASH_SIZE) == 0) {
found_prior = true;
}
}
if (found_prior) {
for (int i = 0; i < MAX_DIRECT_RETRY_SLOTS; i++) {
if (i == replacement_direct_slot || !_direct_retries[i].active
|| !_direct_retries[i].has_message_replacement_key
|| _direct_retries[i].message_timestamp == message_timestamp
|| memcmp(_direct_retries[i].message_replacement_key, message_key,
MAX_HASH_SIZE) != 0) {
continue;
}
retireDirectRetrySlot(i);
}
for (int i = 0; i < MAX_FLOOD_RETRY_SLOTS; i++) {
if (i == replacement_flood_slot || !_flood_retries[i].active
|| !_flood_retries[i].has_message_replacement_key
|| _flood_retries[i].message_timestamp == message_timestamp
|| memcmp(_flood_retries[i].message_replacement_key, message_key,
MAX_HASH_SIZE) != 0) {
continue;
}
retireFloodRetrySlot(i);
}
// A full retry table may have prevented the replacement from reserving a
// slot before it was queued. The matching stale slots are free now.
if (replacement_direct_slot < 0 && replacement_flood_slot < 0) {
if (replacement_packet->isRouteDirect()) {
maybeScheduleDirectRetry(replacement_packet, 0);
} else if (replacement_packet->isRouteFlood()) {
maybeScheduleFloodRetry(replacement_packet, 1);
}
}
}
// Locate a slot registered above when a full table initially rejected the
// new packet. If no slot exists, the role's ordinary retry policy declined
// this message and there is no retry state to tag.
if (replacement_direct_slot < 0 && replacement_flood_slot < 0) {
for (int i = 0; i < MAX_DIRECT_RETRY_SLOTS; i++) {
if (_direct_retries[i].active
&& _direct_retries[i].trigger_packet == replacement_packet) {
replacement_direct_slot = i;
break;
}
}
if (replacement_direct_slot < 0) {
for (int i = 0; i < MAX_FLOOD_RETRY_SLOTS; i++) {
if (_flood_retries[i].active
&& _flood_retries[i].trigger_packet == replacement_packet) {
replacement_flood_slot = i;
break;
}
}
}
}
if (replacement_direct_slot >= 0) {
DirectRetryEntry& entry = _direct_retries[replacement_direct_slot];
memcpy(entry.message_replacement_key, message_key, MAX_HASH_SIZE);
entry.message_timestamp = message_timestamp;
entry.has_message_replacement_key = true;
} else if (replacement_flood_slot >= 0) {
FloodRetryEntry& entry = _flood_retries[replacement_flood_slot];
memcpy(entry.message_replacement_key, message_key, MAX_HASH_SIZE);
entry.message_timestamp = message_timestamp;
entry.has_message_replacement_key = true;
}
}
bool Mesh::hasActiveRetries(const uint8_t retry_key[MAX_HASH_SIZE]) const {
if (retry_key == NULL || (_active_direct_retry_count == 0 && _active_flood_retry_count == 0)) {
return false;
src/Mesh.h
+15
View File
@@ -64,9 +64,11 @@ class Mesh : public Dispatcher {
unsigned long echo_wait_started_at;
unsigned long retry_at;
uint32_t retry_delay;
uint32_t message_timestamp;
uint8_t retry_attempts_sent;
uint8_t retry_key[MAX_HASH_SIZE];
uint8_t trace_replacement_key[MAX_HASH_SIZE];
uint8_t message_replacement_key[MAX_HASH_SIZE];
uint8_t next_hop_hash[MAX_HASH_SIZE];
uint8_t next_hop_hash_len;
uint8_t payload_type;
@@ -76,6 +78,7 @@ class Mesh : public Dispatcher {
bool final_hop_retry;
bool waiting_final_echo;
bool queued;
bool has_message_replacement_key;
bool active;
};
@@ -85,13 +88,16 @@ class Mesh : public Dispatcher {
unsigned long retry_started_at;
unsigned long retry_at;
uint32_t retry_delay;
uint32_t message_timestamp;
uint8_t retry_attempts_sent;
uint8_t retry_key[MAX_HASH_SIZE];
uint8_t message_replacement_key[MAX_HASH_SIZE];
uint8_t priority;
uint8_t progress_marker;
bool self_advert;
bool waiting_final_echo;
bool queued;
bool has_message_replacement_key;
bool active;
};
@@ -244,6 +250,15 @@ protected:
void replaceActiveRetries(const Packet* replacement_packet,
const uint8_t retry_key[MAX_HASH_SIZE]);
/**
* \brief Mark a successfully queued text message as the current retry owner.
* Older retries with the same caller-defined message key and a
* different timestamp are retired across both direct and flood routes.
*/
void replaceActiveMessageRetries(
const Packet* replacement_packet,
const uint8_t message_key[MAX_HASH_SIZE], uint32_t message_timestamp);
// Cancel every queued/future retry of one route type. A packet already on
// air is allowed to finish, but no later retry is armed from it.
void cancelAllDirectRetries();
src/helpers/BaseChatMesh.cpp
+5 -1
View File
@@ -449,7 +449,8 @@ mesh::Packet* BaseChatMesh::composeMsgPacket(const ContactInfo& recipient, uint3
int BaseChatMesh::sendMessage(const ContactInfo& recipient, uint32_t timestamp, uint8_t attempt, const char* text,
uint32_t& expected_ack, uint32_t& est_timeout, uint8_t* packet_hash,
const uint8_t* replace_retry_key) {
const uint8_t* replace_retry_key,
const uint8_t* message_retry_key) {
mesh::Packet* pkt = composeMsgPacket(recipient, timestamp, attempt, text, expected_ack);
if (pkt == NULL) return MSG_SEND_FAILED;
if (packet_hash != NULL) {
@@ -483,6 +484,9 @@ int BaseChatMesh::sendMessage(const ContactInfo& recipient, uint32_t timestamp,
if (replace_retry_key != NULL) {
replaceActiveRetries(pkt, replace_retry_key);
}
if (message_retry_key != NULL) {
replaceActiveMessageRetries(pkt, message_retry_key, timestamp);
}
txt_send_timeout = futureMillis(est_timeout);
return rc;
}
src/helpers/BaseChatMesh.h
+2 -1
View File
@@ -162,7 +162,8 @@ public:
mesh::Packet* createSelfAdvert(const char* name, double lat, double lon);
int sendMessage(const ContactInfo& recipient, uint32_t timestamp, uint8_t attempt, const char* text,
uint32_t& expected_ack, uint32_t& est_timeout, uint8_t* packet_hash = NULL,
const uint8_t* replace_retry_key = NULL);
const uint8_t* replace_retry_key = NULL,
const uint8_t* message_retry_key = NULL);
int sendCommandData(const ContactInfo& recipient, uint32_t timestamp,
uint8_t attempt, const char* text,
uint32_t& est_timeout,
test/test_trace_retry/test_trace_retry.cpp
+146
View File
@@ -98,6 +98,12 @@ public:
onSendComplete(packet);
}
void trackMessageRetry(const mesh::Packet* packet,
const uint8_t message_key[MAX_HASH_SIZE],
uint32_t message_timestamp) {
replaceActiveMessageRetries(packet, message_key, message_timestamp);
}
mesh::DispatcherAction receivePacket(mesh::Packet* packet) {
return onRecvPacket(packet);
}
@@ -320,6 +326,44 @@ static mesh::Packet* makeTrace(TraceTestMesh& node, uint32_t tag, uint32_t auth,
return packet;
}
static mesh::Packet* makeDirectText(TraceTestMesh& node, uint8_t payload_marker,
const uint8_t* route, uint8_t route_len) {
mesh::Packet* packet = node.obtainNewPacket();
EXPECT_NE(packet, nullptr);
if (packet == nullptr) return nullptr;
packet->header = PAYLOAD_TYPE_TXT_MSG << PH_TYPE_SHIFT;
packet->payload_len = 3;
packet->payload[0] = 0xA1;
packet->payload[1] = 0xB2;
packet->payload[2] = payload_marker;
EXPECT_TRUE(node.sendDirect(packet, route, route_len));
return packet;
}
static mesh::Packet* makeFloodText(TraceTestMesh& node, uint8_t payload_marker) {
mesh::Packet* packet = node.obtainNewPacket();
EXPECT_NE(packet, nullptr);
if (packet == nullptr) return nullptr;
packet->header = PAYLOAD_TYPE_TXT_MSG << PH_TYPE_SHIFT;
packet->payload_len = 3;
packet->payload[0] = 0xA1;
packet->payload[1] = 0xB2;
packet->payload[2] = payload_marker;
EXPECT_TRUE(node.sendFlood(packet));
return packet;
}
static void finishCurrentSend(TraceTestMesh& node, TraceTestClock& clock,
TraceTestRadio& radio) {
clock.now++;
node.loop();
ASSERT_TRUE(radio.sending);
radio.complete = true;
clock.now++;
node.loop();
ASSERT_FALSE(radio.sending);
}
static void initSelfAdvert(TraceTestMesh& node, mesh::Packet* packet, uint8_t marker) {
ASSERT_NE(packet, nullptr);
if (packet == nullptr) return;
@@ -706,6 +750,108 @@ TEST(TraceRetry, NewTraceReplacesQueuedRetryButAdvancedOldTraceStillQueues) {
EXPECT_EQ(2, manager.getOutboundTotal());
}
TEST(MessageRetry, DifferentTimestampReplacesQueuedDirectRetry) {
TraceTestClock clock;
TraceTestRTC rtc;
TraceTestRNG rng;
TraceTestRadio radio;
TraceTestTables tables;
StaticPoolPacketManager manager(12);
TraceTestMesh node(radio, clock, rng, rtc, manager, tables);
node.begin();
const uint8_t route[] = {0x11, 0x22};
const uint8_t message_key[MAX_HASH_SIZE] = {
0x10, 0x20, 0x30, 0x40, 0x50, 0x60, 0x70, 0x80
};
mesh::Packet* old_message = makeDirectText(node, 0x01, route, sizeof(route));
ASSERT_NE(old_message, nullptr);
node.trackMessageRetry(old_message, message_key, 100U);
finishCurrentSend(node, clock, radio);
ASSERT_EQ(1, manager.getOutboundTotal());
mesh::Packet* old_retry = manager.getOutboundByIdx(0);
mesh::Packet* new_message = makeDirectText(node, 0x02, route, sizeof(route));
ASSERT_NE(new_message, nullptr);
ASSERT_EQ(2, manager.getOutboundTotal());
node.trackMessageRetry(new_message, message_key, 101U);
ASSERT_EQ(1, manager.getOutboundTotal());
EXPECT_EQ(new_message, manager.getOutboundByIdx(0));
EXPECT_NE(old_retry, manager.getOutboundByIdx(0));
finishCurrentSend(node, clock, radio);
ASSERT_EQ(1, manager.getOutboundTotal());
EXPECT_NE(old_retry, manager.getOutboundByIdx(0));
}
TEST(MessageRetry, SameTimestampKeepsExistingRetrySequence) {
TraceTestClock clock;
TraceTestRTC rtc;
TraceTestRNG rng;
TraceTestRadio radio;
TraceTestTables tables;
StaticPoolPacketManager manager(12);
TraceTestMesh node(radio, clock, rng, rtc, manager, tables);
node.begin();
const uint8_t route[] = {0x31, 0x32};
const uint8_t message_key[MAX_HASH_SIZE] = {
0x81, 0x71, 0x61, 0x51, 0x41, 0x31, 0x21, 0x11
};
mesh::Packet* old_message = makeDirectText(node, 0x11, route, sizeof(route));
ASSERT_NE(old_message, nullptr);
node.trackMessageRetry(old_message, message_key, 200U);
finishCurrentSend(node, clock, radio);
ASSERT_EQ(1, manager.getOutboundTotal());
mesh::Packet* old_retry = manager.getOutboundByIdx(0);
mesh::Packet* same_timestamp = makeDirectText(node, 0x12, route, sizeof(route));
ASSERT_NE(same_timestamp, nullptr);
node.trackMessageRetry(same_timestamp, message_key, 200U);
ASSERT_EQ(2, manager.getOutboundTotal());
bool found_old_retry = false;
bool found_new_message = false;
for (int i = 0; i < manager.getOutboundTotal(); i++) {
found_old_retry |= manager.getOutboundByIdx(i) == old_retry;
found_new_message |= manager.getOutboundByIdx(i) == same_timestamp;
}
EXPECT_TRUE(found_old_retry);
EXPECT_TRUE(found_new_message);
}
TEST(MessageRetry, ReplacementWorksAcrossFloodAndDirectRoutes) {
TraceTestClock clock;
TraceTestRTC rtc;
TraceTestRNG rng;
TraceTestRadio radio;
TraceTestTables tables;
StaticPoolPacketManager manager(12);
TraceTestMesh node(radio, clock, rng, rtc, manager, tables);
node.begin();
const uint8_t message_key[MAX_HASH_SIZE] = {
0x08, 0x18, 0x28, 0x38, 0x48, 0x58, 0x68, 0x78
};
mesh::Packet* old_flood = makeFloodText(node, 0x21);
ASSERT_NE(old_flood, nullptr);
node.trackMessageRetry(old_flood, message_key, 300U);
finishCurrentSend(node, clock, radio);
ASSERT_EQ(1, manager.getOutboundTotal());
mesh::Packet* old_retry = manager.getOutboundByIdx(0);
const uint8_t route[] = {0x41, 0x42};
mesh::Packet* new_direct = makeDirectText(node, 0x22, route, sizeof(route));
ASSERT_NE(new_direct, nullptr);
ASSERT_EQ(2, manager.getOutboundTotal());
node.trackMessageRetry(new_direct, message_key, 301U);
ASSERT_EQ(1, manager.getOutboundTotal());
EXPECT_EQ(new_direct, manager.getOutboundByIdx(0));
EXPECT_NE(old_retry, manager.getOutboundByIdx(0));
}
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();