From 26d6e0f99ac4f1e5355d860cfa90351985d4a79e Mon Sep 17 00:00:00 2001 From: mikecarper Date: Fri, 7 Aug 2026 23:22:31 -0700 Subject: [PATCH] Harden message retry replacement and seed testing channel --- examples/companion_radio/MyMesh.cpp | 54 ++++++-- examples/companion_radio/MyMesh.h | 7 +- examples/simple_room_server/MyMesh.cpp | 40 ++++-- examples/simple_room_server/MyMesh.h | 3 +- src/Mesh.cpp | 123 +++++++++++++++++ src/Mesh.h | 15 +++ src/helpers/BaseChatMesh.cpp | 6 +- src/helpers/BaseChatMesh.h | 3 +- test/test_trace_retry/test_trace_retry.cpp | 146 +++++++++++++++++++++ 9 files changed, 369 insertions(+), 28 deletions(-) diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index aa2b4bf7e..e3b11488d 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -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"); } diff --git a/examples/companion_radio/MyMesh.h b/examples/companion_radio/MyMesh.h index e2a7086c3..5691ef0bc 100644 --- a/examples/companion_radio/MyMesh.h +++ b/examples/companion_radio/MyMesh.h @@ -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 diff --git a/examples/simple_room_server/MyMesh.cpp b/examples/simple_room_server/MyMesh.cpp index 63ed0e3be..fed75b375 100644 --- a/examples/simple_room_server/MyMesh.cpp +++ b/examples/simple_room_server/MyMesh.cpp @@ -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); } } diff --git a/examples/simple_room_server/MyMesh.h b/examples/simple_room_server/MyMesh.h index b68d92a89..48a4c0e5a 100644 --- a/examples/simple_room_server/MyMesh.h +++ b/examples/simple_room_server/MyMesh.h @@ -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; diff --git a/src/Mesh.cpp b/src/Mesh.cpp index 0ccf9cb1b..e6687712a 100644 --- a/src/Mesh.cpp +++ b/src/Mesh.cpp @@ -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; diff --git a/src/Mesh.h b/src/Mesh.h index ad43fc168..f220c08ac 100644 --- a/src/Mesh.h +++ b/src/Mesh.h @@ -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(); diff --git a/src/helpers/BaseChatMesh.cpp b/src/helpers/BaseChatMesh.cpp index d9d3fe168..09f867afb 100644 --- a/src/helpers/BaseChatMesh.cpp +++ b/src/helpers/BaseChatMesh.cpp @@ -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; } diff --git a/src/helpers/BaseChatMesh.h b/src/helpers/BaseChatMesh.h index 39852c0f4..fe4b74b40 100644 --- a/src/helpers/BaseChatMesh.h +++ b/src/helpers/BaseChatMesh.h @@ -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, diff --git a/test/test_trace_retry/test_trace_retry.cpp b/test/test_trace_retry/test_trace_retry.cpp index 0b855077c..6ab9a0460 100644 --- a/test/test_trace_retry/test_trace_retry.cpp +++ b/test/test_trace_retry/test_trace_retry.cpp @@ -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();