From 04d5f335752c4bbf6cc15b64a0444570b9cdaf40 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Sat, 1 Aug 2026 17:13:03 -0700 Subject: [PATCH] Retry messages until an echo is heard --- src/MyMesh.cpp | 616 ++++++++++++++++++++++++++- src/MyMesh.h | 85 +++- src/helpers/CompanionRetryPolicy.h | 64 +++ src/ui-touch/TouchSleep.cpp | 21 +- src/ui-touch/TouchSleep.h | 2 +- src/ui-touch/UITask.cpp | 21 +- test/test_companion_retry_policy.cpp | 55 +++ 7 files changed, 826 insertions(+), 38 deletions(-) create mode 100644 src/helpers/CompanionRetryPolicy.h create mode 100644 test/test_companion_retry_policy.cpp diff --git a/src/MyMesh.cpp b/src/MyMesh.cpp index 621c690..5d2be6f 100644 --- a/src/MyMesh.cpp +++ b/src/MyMesh.cpp @@ -13,6 +13,7 @@ #endif #endif #include +#include "helpers/CompanionRetryPolicy.h" #include #include #include "WiFiConfig.h" @@ -1391,7 +1392,500 @@ static inline bool isMsgFloodType(uint8_t t) { return t == PAYLOAD_TYPE_TXT_MSG || t == PAYLOAD_TYPE_GRP_TXT; } +uint8_t MyMesh::companionDetachQueuedText(mesh::Packet* packets[], uint8_t priorities[], + uint32_t scheduled_for[]) { + uint8_t count = 0; + int queue_idx = 0; + const uint32_t now = _ms->getMillis(); + while (queue_idx < _mgr->getOutboundTotal() + && count < COMPANION_TEXT_QUEUE_CAPACITY) { + mesh::Packet* packet = _mgr->getOutboundByIdx(queue_idx); + if (!packet || packet->getPayloadType() != PAYLOAD_TYPE_TXT_MSG) { + queue_idx++; + continue; + } + + packet = _mgr->removeOutboundByIdx(queue_idx); + if (!packet) continue; + + packets[count] = packet; + priorities[count] = packet->isRouteDirect() ? 0 : 1; + scheduled_for[count] = now; + for (int i = 0; i < COMPANION_RETRY_SLOTS; i++) { + const CompanionRetrySlot& slot = _companion_retries[i]; + if (slot.active && slot.queued_packet == packet) { + priorities[count] = slot.priority; + scheduled_for[count] = slot.retry_at; + break; + } + } + count++; + } + return count; +} + +bool MyMesh::companionRestoreQueuedText(mesh::Packet* packets[], const uint8_t priorities[], + const uint32_t scheduled_for[], uint8_t count) { + bool restored_all = true; + for (uint8_t i = 0; i < count; i++) { + mesh::Packet* packet = packets[i]; + if (!packet) continue; + + _mgr->queueOutbound(packet, priorities[i], scheduled_for[i]); + bool found = false; + for (int j = 0; j < _mgr->getOutboundTotal(); j++) { + if (_mgr->getOutboundByIdx(j) == packet) { + found = true; + break; + } + } + if (found) continue; + + // queueOutbound() returned a rejected packet to the pool. Retire any + // retry metadata that still referred to that pool object. + restored_all = false; + for (int j = 0; j < COMPANION_RETRY_SLOTS; j++) { + if (_companion_retries[j].active + && _companion_retries[j].queued_packet == packet) { + companionRetryResetSlot(j); + } + } + } + return restored_all; +} + +mesh::Packet* MyMesh::companionDetachQueuedTextByHash4( + uint32_t packet_hash4, uint8_t retry_key[MAX_HASH_SIZE]) { + for (int i = 0; i < _mgr->getOutboundTotal(); i++) { + mesh::Packet* packet = _mgr->getOutboundByIdx(i); + if (!packet || packet->getPayloadType() != PAYLOAD_TYPE_TXT_MSG) continue; + + uint8_t candidate_key[MAX_HASH_SIZE]; + uint32_t candidate_hash4 = 0; + packet->calculatePacketHash(candidate_key); + memcpy(&candidate_hash4, candidate_key, sizeof(candidate_hash4)); + if (candidate_hash4 != packet_hash4) continue; + + memcpy(retry_key, candidate_key, MAX_HASH_SIZE); + return _mgr->removeOutboundByIdx(i); + } + return nullptr; +} + +int MyMesh::sendMessage(const ContactInfo& recipient, uint32_t timestamp, uint8_t attempt, + const char* text, uint32_t& expected_ack, uint32_t& est_timeout, + uint32_t* out_packet_hash4, TxtTxDebugInfo* out_dbg) { + if (!text) { + expected_ack = 0; + est_timeout = 0; + return MSG_SEND_FAILED; + } + + mesh::Packet* held[COMPANION_TEXT_QUEUE_CAPACITY] = {}; + uint8_t held_priorities[COMPANION_TEXT_QUEUE_CAPACITY] = {}; + uint32_t held_schedules[COMPANION_TEXT_QUEUE_CAPACITY] = {}; + const uint8_t held_count = companionDetachQueuedText( + held, held_priorities, held_schedules); + + _last_plain_tx_meta_valid = false; + uint32_t packet_hash4 = 0; + int result = BaseChatMesh::sendMessage(recipient, timestamp, attempt, text, + expected_ack, est_timeout, + &packet_hash4, out_dbg); + + uint8_t retry_key[MAX_HASH_SIZE] = {}; + mesh::Packet* newest = result == MSG_SEND_FAILED + ? nullptr + : companionDetachQueuedTextByHash4(packet_hash4, retry_key); + + const bool restored_old = companionRestoreQueuedText( + held, held_priorities, held_schedules, held_count); + bool restored_new = newest != nullptr; + if (newest) { + mesh::Packet* one[] = {newest}; + const uint8_t priority[] = {static_cast(newest->isRouteDirect() ? 0 : 1)}; + const uint32_t scheduled[] = {_ms->getMillis()}; + restored_new = companionRestoreQueuedText(one, priority, scheduled, 1); + } + + if (!restored_old) { + MESH_DEBUG_PRINTLN("%s MyMesh::sendMessage(): failed to restore queued TXT", getLogDateTime()); + } + if (result != MSG_SEND_FAILED && !restored_new) { + expected_ack = 0; + est_timeout = 0; + result = MSG_SEND_FAILED; + } + + if (out_packet_hash4) *out_packet_hash4 = packet_hash4; + if (result != MSG_SEND_FAILED) { + _last_plain_tx_ack = expected_ack; + memcpy(_last_plain_tx_retry_key, retry_key, sizeof(_last_plain_tx_retry_key)); + mesh::Utils::sha256(_last_plain_tx_fingerprint, + sizeof(_last_plain_tx_fingerprint), + recipient.id.pub_key, PUB_KEY_SIZE, + reinterpret_cast(text), strlen(text)); + _last_plain_tx_meta_valid = true; + } + return result; +} + +int MyMesh::sendCommandData(const ContactInfo& recipient, uint32_t timestamp, + uint8_t attempt, const char* text, uint32_t& est_timeout, + uint32_t* out_packet_hash4, TxtTxDebugInfo* out_dbg) { + if (!text) { + est_timeout = 0; + return MSG_SEND_FAILED; + } + + mesh::Packet* held[COMPANION_TEXT_QUEUE_CAPACITY] = {}; + uint8_t held_priorities[COMPANION_TEXT_QUEUE_CAPACITY] = {}; + uint32_t held_schedules[COMPANION_TEXT_QUEUE_CAPACITY] = {}; + const uint8_t held_count = companionDetachQueuedText( + held, held_priorities, held_schedules); + + _last_plain_tx_meta_valid = false; + uint32_t packet_hash4 = 0; + int result = BaseChatMesh::sendCommandData(recipient, timestamp, attempt, text, + est_timeout, &packet_hash4, out_dbg); + + uint8_t retry_key[MAX_HASH_SIZE] = {}; + mesh::Packet* newest = result == MSG_SEND_FAILED + ? nullptr + : companionDetachQueuedTextByHash4(packet_hash4, retry_key); + + const bool restored_old = companionRestoreQueuedText( + held, held_priorities, held_schedules, held_count); + bool restored_new = newest != nullptr; + if (newest) { + mesh::Packet* one[] = {newest}; + const uint8_t priority[] = {static_cast(newest->isRouteDirect() ? 0 : 1)}; + const uint32_t scheduled[] = {_ms->getMillis()}; + restored_new = companionRestoreQueuedText(one, priority, scheduled, 1); + } + + if (!restored_old) { + MESH_DEBUG_PRINTLN("%s MyMesh::sendCommandData(): failed to restore queued TXT", getLogDateTime()); + } + if (result != MSG_SEND_FAILED && !restored_new) { + est_timeout = 0; + result = MSG_SEND_FAILED; + } + if (out_packet_hash4) *out_packet_hash4 = packet_hash4; + return result; +} + +uint32_t MyMesh::companionRetryDelay(const mesh::Packet* packet, bool direct, + uint8_t attempt_idx) { + if (!packet || !_radio) return 0; + + const uint32_t packet_airtime = _radio->getEstAirtimeFor(packet->getRawLength()); + if (direct) { + return CompanionRetryPolicy::directDelay(packet_airtime, attempt_idx); + } + + const uint32_t max_packet_airtime = _radio->getEstAirtimeFor(MAX_TRANS_UNIT); + const uint32_t jitter_percent = getRNG()->nextInt(0, 201); + return CompanionRetryPolicy::floodDelay(max_packet_airtime, packet_airtime, + jitter_percent); +} + +void MyMesh::companionRetryResetSlot(int slot_idx) { + if (slot_idx < 0 || slot_idx >= COMPANION_RETRY_SLOTS) return; + + CompanionRetrySlot& slot = _companion_retries[slot_idx]; + if (slot.active && _active_companion_retries > 0) { + _active_companion_retries--; + } + slot.queued_packet = nullptr; + slot.retry_at = 0; + slot.retry_delay = 0; + slot.missing_since = 0; + memset(slot.retry_key, 0, sizeof(slot.retry_key)); + slot.attempts_sent = 0; + slot.max_attempts = 0; + slot.priority = 0; + slot.progress_marker = 0; + slot.payload_type = 0; + slot.direct = false; + slot.waiting_final_echo = false; + slot.active = false; +} + +void MyMesh::companionRetryCancelSlot(int slot_idx) { + if (slot_idx < 0 || slot_idx >= COMPANION_RETRY_SLOTS) return; + + CompanionRetrySlot& slot = _companion_retries[slot_idx]; + mesh::Packet* queued = slot.queued_packet; + if (slot.active && queued) { + for (int i = 0; i < _mgr->getOutboundTotal(); i++) { + if (_mgr->getOutboundByIdx(i) != queued) continue; + + uint8_t queued_key[MAX_HASH_SIZE]; + queued->calculatePacketHash(queued_key); + if (memcmp(queued_key, slot.retry_key, sizeof(queued_key)) == 0) { + mesh::Packet* removed = _mgr->removeOutboundByIdx(i); + if (removed) releasePacket(removed); + } + break; + } + } + companionRetryResetSlot(slot_idx); +} + +void MyMesh::companionRetryCancelKey(const uint8_t retry_key[MAX_HASH_SIZE]) { + if (!CompanionRetryPolicy::keyIsSet(retry_key, MAX_HASH_SIZE)) return; + + // First retire tracked retry trains. companionRetryCancelSlot() also removes + // their queued clone when it has not entered the radio yet. + for (int i = 0; i < COMPANION_RETRY_SLOTS; i++) { + const CompanionRetrySlot& slot = _companion_retries[i]; + if (slot.active && CompanionRetryPolicy::keysEqual( + slot.retry_key, retry_key, MAX_HASH_SIZE)) { + companionRetryCancelSlot(i); + } + } + + // A semantic replacement can arrive before the original packet's first TX, + // before a CompanionRetrySlot exists. Remove that exact queued packet too. + for (int i = _mgr->getOutboundTotal() - 1; i >= 0; i--) { + mesh::Packet* packet = _mgr->getOutboundByIdx(i); + if (!packet || packet->getPayloadType() != PAYLOAD_TYPE_TXT_MSG) continue; + + uint8_t queued_key[MAX_HASH_SIZE]; + packet->calculatePacketHash(queued_key); + if (!CompanionRetryPolicy::keysEqual( + queued_key, retry_key, MAX_HASH_SIZE)) continue; + + mesh::Packet* removed = _mgr->removeOutboundByIdx(i); + if (removed) releasePacket(removed); + } +} + +bool MyMesh::companionRetryQueueClone(int slot_idx, const mesh::Packet* packet, + uint8_t attempt_idx) { + if (slot_idx < 0 || slot_idx >= COMPANION_RETRY_SLOTS || !packet) return false; + + CompanionRetrySlot& slot = _companion_retries[slot_idx]; + mesh::Packet* retry = obtainNewPacket(); + if (!retry) return false; + + *retry = *packet; // exact duplicate: timestamp and ciphertext stay fixed + slot.retry_delay = companionRetryDelay(packet, slot.direct, attempt_idx); + slot.retry_at = _ms->getMillis() + slot.retry_delay; + _mgr->queueOutbound(retry, slot.priority, slot.retry_at); + + for (int i = 0; i < _mgr->getOutboundTotal(); i++) { + if (_mgr->getOutboundByIdx(i) == retry) { + slot.queued_packet = retry; + slot.missing_since = 0; + return true; + } + } + + // StaticPoolPacketManager already returned a rejected packet to its pool. + return false; +} + +void MyMesh::companionRetryStart(const mesh::Packet* packet, + const uint8_t retry_key[MAX_HASH_SIZE]) { + if (!packet || !retry_key) return; + + const uint8_t payload_type = packet->getPayloadType(); + const uint8_t path_count = packet->getPathHashCount(); + const bool text_from_self = packet->payload_len >= 2U * PATH_HASH_SIZE + && self_id.isHashMatch(&packet->payload[PATH_HASH_SIZE], PATH_HASH_SIZE); + + bool direct = false; + uint8_t max_attempts = 0; + uint8_t priority = 0; + if (packet->isRouteDirect() && payload_type == PAYLOAD_TYPE_TXT_MSG + && path_count > 0 && text_from_self) { + direct = true; + max_attempts = CompanionRetryPolicy::DIRECT_MAX_ATTEMPTS; + } else if (packet->isRouteFlood() && path_count == 0 + && ((payload_type == PAYLOAD_TYPE_TXT_MSG && text_from_self) + || payload_type == PAYLOAD_TYPE_GRP_TXT)) { + max_attempts = CompanionRetryPolicy::FLOOD_MAX_ATTEMPTS; + priority = 1; + } else { + return; + } + + for (int i = 0; i < COMPANION_RETRY_SLOTS; i++) { + const CompanionRetrySlot& slot = _companion_retries[i]; + if (slot.active && slot.direct == direct + && memcmp(slot.retry_key, retry_key, MAX_HASH_SIZE) == 0) { + return; + } + } + + int slot_idx = -1; + for (int i = 0; i < COMPANION_RETRY_SLOTS; i++) { + if (!_companion_retries[i].active) { + slot_idx = i; + break; + } + } + if (slot_idx < 0) return; + + CompanionRetrySlot& slot = _companion_retries[slot_idx]; + slot.queued_packet = nullptr; + slot.missing_since = 0; + memcpy(slot.retry_key, retry_key, MAX_HASH_SIZE); + slot.attempts_sent = 0; + slot.max_attempts = max_attempts; + slot.priority = priority; + slot.progress_marker = path_count; + slot.payload_type = payload_type; + slot.direct = direct; + slot.waiting_final_echo = false; + slot.active = true; + _active_companion_retries++; + if (!companionRetryQueueClone(slot_idx, packet, 0)) { + companionRetryResetSlot(slot_idx); + } +} + +void MyMesh::logTx(mesh::Packet* packet, int len) { + (void)len; + if (!packet) return; + + uint8_t packet_key[MAX_HASH_SIZE]; + packet->calculatePacketHash(packet_key); + + for (int i = 0; i < COMPANION_RETRY_SLOTS; i++) { + CompanionRetrySlot& slot = _companion_retries[i]; + if (!slot.active || slot.queued_packet != packet) continue; + + // If a pool object was unexpectedly reused, do not mistake the new packet + // for this retry. + if (memcmp(slot.retry_key, packet_key, MAX_HASH_SIZE) != 0) { + companionRetryResetSlot(i); + break; + } + + slot.queued_packet = nullptr; + slot.missing_since = 0; + slot.attempts_sent++; + if (slot.attempts_sent >= slot.max_attempts) { + // Keep the metadata for one last echo window. Dispatcher releases the + // just-transmitted pool packet after this hook returns. + slot.waiting_final_echo = true; + slot.retry_at = _ms->getMillis() + slot.retry_delay; + } else if (!companionRetryQueueClone(i, packet, slot.attempts_sent)) { + companionRetryResetSlot(i); + } + return; + } + + companionRetryStart(packet, packet_key); +} + +void MyMesh::logTxFail(mesh::Packet* packet, int len) { + (void)len; + if (!packet) return; + + for (int i = 0; i < COMPANION_RETRY_SLOTS; i++) { + CompanionRetrySlot& slot = _companion_retries[i]; + if (!slot.active || slot.queued_packet != packet) continue; + + // Dispatcher owns and releases this in-flight packet after the hook. + slot.queued_packet = nullptr; + companionRetryResetSlot(i); + return; + } +} + +void MyMesh::companionRetryObserveRaw(const uint8_t raw[], int len) { + if (_active_companion_retries == 0 || !raw || len <= 0) return; + + const uint8_t payload_type = (raw[0] >> PH_TYPE_SHIFT) & PH_TYPE_MASK; + if (payload_type != PAYLOAD_TYPE_TXT_MSG && payload_type != PAYLOAD_TYPE_GRP_TXT) { + return; + } + + mesh::Packet packet; + if (!tryParsePacket(&packet, raw, len)) return; + + const bool direct = packet.isRouteDirect(); + if ((!direct && !packet.isRouteFlood()) + || (direct && payload_type != PAYLOAD_TYPE_TXT_MSG)) { + return; + } + + uint8_t retry_key[MAX_HASH_SIZE]; + packet.calculatePacketHash(retry_key); + const uint8_t received_path_count = packet.getPathHashCount(); + for (int i = 0; i < COMPANION_RETRY_SLOTS; i++) { + const CompanionRetrySlot& slot = _companion_retries[i]; + if (!slot.active || slot.direct != direct + || memcmp(slot.retry_key, retry_key, MAX_HASH_SIZE) != 0) { + continue; + } + + const bool is_echo = direct + ? CompanionRetryPolicy::isDirectEcho(slot.progress_marker, received_path_count) + : CompanionRetryPolicy::isFloodEcho(slot.progress_marker, received_path_count); + if (is_echo) companionRetryCancelSlot(i); + } +} + +void MyMesh::companionRetryService() { + if (_active_companion_retries == 0) return; + + const uint32_t now = _ms->getMillis(); + const uint32_t missing_grace = 1000UL + + (2UL * _radio->getEstAirtimeFor(MAX_TRANS_UNIT)); + for (int i = 0; i < COMPANION_RETRY_SLOTS; i++) { + CompanionRetrySlot& slot = _companion_retries[i]; + if (!slot.active) continue; + + if (slot.waiting_final_echo) { + if (millisHasNowPassed(slot.retry_at)) companionRetryResetSlot(i); + continue; + } + + if (slot.queued_packet) { + bool found = false; + bool hash_matches = false; + for (int j = 0; j < _mgr->getOutboundTotal(); j++) { + mesh::Packet* queued = _mgr->getOutboundByIdx(j); + if (queued != slot.queued_packet) continue; + + uint8_t queued_key[MAX_HASH_SIZE]; + queued->calculatePacketHash(queued_key); + found = true; + hash_matches = memcmp(queued_key, slot.retry_key, MAX_HASH_SIZE) == 0; + break; + } + + if (found) { + if (!hash_matches) { + // The pool object was reused after another send removed our retry. + companionRetryResetSlot(i); + } else { + slot.missing_since = 0; + } + } else if (slot.missing_since == 0) { + // Usually this means Dispatcher moved the packet from the queue to the + // radio. Allow enough time for its normal TX-complete/fail callback. + slot.missing_since = now == 0 ? 1 : now; + } else if ((uint32_t)(now - slot.missing_since) > missing_grace) { + // The pool object is no longer ours, so only retire the metadata here. + companionRetryResetSlot(i); + } + continue; + } + + // A non-final active slot always owns one queued or in-flight clone. + companionRetryResetSlot(i); + } +} + void MyMesh::logRxRaw(float snr, float rssi, const uint8_t raw[], int len) { + companionRetryObserveRaw(raw, len); + const int8_t snr_q4 = (int8_t)(snr * 4.0f); const uint32_t now_ms = millis(); // Parse route + path length (hop count). Header byte layout is @@ -1844,6 +2338,74 @@ void MyMesh::onContactPathUpdated(const ContactInfo &contact) { dirty_contacts_expiry = futureMillis(LAZY_CONTACTS_WRITE_DELAY); } +void MyMesh::clearExpectedAck(AckTableEntry& entry, bool cancel_retry) { + if (cancel_retry) companionRetryCancelKey(entry.retry_key); + memset(&entry, 0, sizeof(entry)); +} + +MyMesh::AckTableEntry* MyMesh::findPendingTextMessage( + const uint8_t text_fingerprint[MAX_HASH_SIZE]) { + if (!text_fingerprint) return nullptr; + for (int i = 0; i < EXPECTED_ACK_TABLE_SIZE; i++) { + AckTableEntry& entry = expected_ack_table[i]; + if (CompanionRetryPolicy::shouldReplacePendingText( + entry.ack, entry.text_fingerprint, text_fingerprint, MAX_HASH_SIZE)) { + return &entry; + } + } + return nullptr; +} + +void MyMesh::uiRegisterExpectedAck(uint32_t expected_ack, const uint8_t pub_key[32]) { + if (expected_ack == 0 || !pub_key) return; + + // Transport retries of the same command frame reuse the ACK. Do not create + // another table entry or disturb the packet's existing retry ownership. + for (int i = 0; i < EXPECTED_ACK_TABLE_SIZE; i++) { + if (expected_ack_table[i].ack == expected_ack) { + _last_plain_tx_meta_valid = false; + return; + } + } + + ContactInfo* contact = lookupContactByPubKey(pub_key, PUB_KEY_SIZE); + if (!contact) { + _last_plain_tx_meta_valid = false; + return; + } + + const bool has_send_meta = _last_plain_tx_meta_valid + && _last_plain_tx_ack == expected_ack; + AckTableEntry* replacement = has_send_meta + ? findPendingTextMessage(_last_plain_tx_fingerprint) + : nullptr; + + AckTableEntry* entry; + if (replacement) { + // Only the same recipient+text supersedes an older pending message. + // Unrelated messages retain both their ACK record and retry train. + clearExpectedAck(*replacement, true); + entry = replacement; + } else { + entry = &expected_ack_table[next_ack_idx]; + // Circular-table eviction is bookkeeping only: do not cancel an unrelated + // message's lower-level retries merely because its ACK slot is reused. + clearExpectedAck(*entry, false); + next_ack_idx = (next_ack_idx + 1) % EXPECTED_ACK_TABLE_SIZE; + } + + entry->msg_sent = _ms->getMillis(); + entry->ack = expected_ack; + entry->contact = contact; + if (has_send_meta) { + memcpy(entry->text_fingerprint, _last_plain_tx_fingerprint, + sizeof(entry->text_fingerprint)); + memcpy(entry->retry_key, _last_plain_tx_retry_key, + sizeof(entry->retry_key)); + } + _last_plain_tx_meta_valid = false; +} + ContactInfo* MyMesh::processAck(const uint8_t *data) { #if defined(DISPLAY_CLASS) // Diag: log every processAck call so we can see whether the ACK matching @@ -1858,7 +2420,7 @@ ContactInfo* MyMesh::processAck(const uint8_t *data) { #endif // see if matches any in a table for (int i = 0; i < EXPECTED_ACK_TABLE_SIZE; i++) { - if (memcmp(data, &expected_ack_table[i].ack, 4) == 0) { // got an ACK from recipient + if (CompanionRetryPolicy::ackMatches(expected_ack_table[i].ack, data)) { out_frame[0] = PUSH_CODE_SEND_CONFIRMED; memcpy(&out_frame[1], data, 4); uint32_t trip_time = _ms->getMillis() - expected_ack_table[i].msg_sent; @@ -1874,9 +2436,11 @@ ContactInfo* MyMesh::processAck(const uint8_t *data) { } #endif - // NOTE: the same ACK can be received multiple times! - expected_ack_table[i].ack = 0; // clear expected hash, now that we have received ACK - return expected_ack_table[i].contact; + // An ACK is stronger delivery evidence than a local repeater echo. Stop + // the exact retry train before clearing its bookkeeping entry. + ContactInfo* contact = expected_ack_table[i].contact; + clearExpectedAck(expected_ack_table[i], true); + return contact; } } return checkConnectionsAck(data); @@ -2589,6 +3153,7 @@ MyMesh::MyMesh(mesh::Radio &radio, mesh::RNG &rng, mesh::RTCClock &rtc, SimpleMe app_target_ver = 0; clearPendingReqs(); _ui_pending_status = 0; + memset(expected_ack_table, 0, sizeof(expected_ack_table)); next_ack_idx = 0; sign_data = NULL; dirty_contacts_expiry = 0; @@ -3021,7 +3586,6 @@ void MyMesh::handleCmdFrame(size_t len) { _ui->appendDiag(line); } } - // TODO: add expected ACK to table if (result == MSG_SEND_FAILED) { writeErrFrame(ERR_CODE_TABLE_FULL); } else { @@ -3030,10 +3594,7 @@ void MyMesh::handleCmdFrame(size_t len) { s_last_cmd_txt_est_timeout = est_timeout; } if (expected_ack) { - expected_ack_table[next_ack_idx].msg_sent = _ms->getMillis(); // add to circular table - expected_ack_table[next_ack_idx].ack = expected_ack; - expected_ack_table[next_ack_idx].contact = recipient; - next_ack_idx = (next_ack_idx + 1) % EXPECTED_ACK_TABLE_SIZE; + uiRegisterExpectedAck(expected_ack, recipient->id.pub_key); } out_frame[0] = RESP_CODE_SENT; @@ -4496,6 +5057,7 @@ void MyMesh::checkSerialInterface() { void MyMesh::loop() { BaseChatMesh::loop(); + companionRetryService(); // Session keep-alives for logged-in servers (rooms). The core pinger sends the // 9-byte REQ_TYPE_KEEP_ALIVE (+ our sync_since) a room server expects; the ACK @@ -4662,7 +5224,37 @@ bool MyMesh::sendAdvert(bool flood) { return true; } -// To check if there is pending work -bool MyMesh::hasPendingWork() const { - return _mgr->getOutboundTotal() > 0 || dirty_contacts_expiry != 0; +bool MyMesh::getNextCompanionRetryWakeDelay(uint32_t& delay_millis) const { + if (_active_companion_retries == 0) return false; + + const uint32_t now = _ms->getMillis(); + bool found = false; + uint32_t shortest_delay = 0; + for (int i = 0; i < COMPANION_RETRY_SLOTS; i++) { + const CompanionRetrySlot& slot = _companion_retries[i]; + if (!slot.active) continue; + + const uint32_t candidate = CompanionRetryPolicy::wakeDelay(now, slot.retry_at); + if (!found || candidate < shortest_delay) { + shortest_delay = candidate; + found = true; + } + } + if (found) delay_millis = shortest_delay; + return found; +} + +// Future queue entries, retry echo windows, and contact-write timers are wake +// deadlines; they should not prevent the MCU's idle power-saving path. Report +// only work that is due now. An in-flight retry keeps retry_at in the past until +// its TX callback runs, so the CPU remains awake while the radio is transmitting. +bool MyMesh::hasPendingWork() const { + const uint32_t now = _ms->getMillis(); + if (_mgr->getOutboundCount(now) > 0) return true; + if (dirty_contacts_expiry != 0 && millisHasNowPassed(dirty_contacts_expiry)) { + return true; + } + + uint32_t retry_delay = 0; + return getNextCompanionRetryWakeDelay(retry_delay) && retry_delay == 0; } diff --git a/src/MyMesh.h b/src/MyMesh.h index d1c65c8..a881059 100644 --- a/src/MyMesh.h +++ b/src/MyMesh.h @@ -130,6 +130,17 @@ public: void begin(bool has_display); void startInterface(BaseSerialInterface &serial); + // Keep WadaMesh's queued text packets intact around the pinned core's + // sendMessage/sendCommandData implementations. That core currently drops + // every queued TXT before enqueueing a new one; these wrappers restore FIFO + // behavior and let MyMesh replace only the same logical private message. + int sendMessage(const ContactInfo& recipient, uint32_t timestamp, uint8_t attempt, + const char* text, uint32_t& expected_ack, uint32_t& est_timeout, + uint32_t* out_packet_hash4 = nullptr, TxtTxDebugInfo* out_dbg = nullptr); + int sendCommandData(const ContactInfo& recipient, uint32_t timestamp, uint8_t attempt, + const char* text, uint32_t& est_timeout, + uint32_t* out_packet_hash4 = nullptr, TxtTxDebugInfo* out_dbg = nullptr); + const char *getNodeName(); NodePrefs *getNodePrefs(); uint32_t getBLEPin(); @@ -245,6 +256,8 @@ protected: void sendFloodScoped(const mesh::GroupChannel& channel, mesh::Packet* pkt, uint32_t delay_millis=0) override; void logRxRaw(float snr, float rssi, const uint8_t raw[], int len) override; + void logTx(mesh::Packet* packet, int len) override; + void logTxFail(mesh::Packet* packet, int len) override; bool isAutoAddEnabled() const override; bool shouldAutoAddContactType(uint8_t type) const override; bool shouldOverwriteWhenFull() const override; @@ -434,18 +447,7 @@ public: * onMessageAcked back to the UI. The companion-serial CMD_SEND_TXT_MSG * handler already does this for app-originated messages; the touch UI * reaches sendMessage directly and skipped this until now. */ - void uiRegisterExpectedAck(uint32_t expected_ack, const uint8_t pub_key[32]) { - if (expected_ack == 0) return; - ContactInfo* c = lookupContactByPubKey(pub_key, PUB_KEY_SIZE); - if (!c) return; - expected_ack_table[next_ack_idx].msg_sent = _ms->getMillis(); - expected_ack_table[next_ack_idx].ack = expected_ack; - expected_ack_table[next_ack_idx].contact = c; - // EXPECTED_ACK_TABLE_SIZE is #defined further down in this header next - // to the table itself; hard-code 8 here so this inline helper compiles - // wherever it's used. - next_ack_idx = (next_ack_idx + 1) % 8; - } + void uiRegisterExpectedAck(uint32_t expected_ack, const uint8_t pub_key[32]); // ---- "Repeats heard" for sent floods + route of the last received flood ---- // When we originate a flood TXT, repeaters re-broadcast it and our own radio @@ -970,6 +972,11 @@ public: // To check if there is pending work bool hasPendingWork() const; + /** Return the delay until the next companion retry/final-echo deadline. + * Future retries are wake deadlines, not work that should hold the CPU + * awake. Returns false when no retry train is active. */ + bool getNextCompanionRetryWakeDelay(uint32_t& delay_millis) const; + // Number of companion clients currently connected on any transport. // Used by the idle light-sleep gate (TouchSleep) to confirm no one is // actively talking to us before the node parks in light sleep. @@ -981,6 +988,24 @@ public: bool isRadioReceiving() const { return _radio && _radio->isReceiving(); } private: + static const uint8_t COMPANION_TEXT_QUEUE_CAPACITY = 16; + uint8_t companionDetachQueuedText(mesh::Packet* packets[], uint8_t priorities[], + uint32_t scheduled_for[]); + bool companionRestoreQueuedText(mesh::Packet* packets[], const uint8_t priorities[], + const uint32_t scheduled_for[], uint8_t count); + mesh::Packet* companionDetachQueuedTextByHash4(uint32_t packet_hash4, + uint8_t retry_key[MAX_HASH_SIZE]); + void companionRetryCancelKey(const uint8_t retry_key[MAX_HASH_SIZE]); + + void companionRetryObserveRaw(const uint8_t raw[], int len); + void companionRetryStart(const mesh::Packet* packet, const uint8_t retry_key[MAX_HASH_SIZE]); + void companionRetryService(); + void companionRetryCancelSlot(int slot_idx); + void companionRetryResetSlot(int slot_idx); + bool companionRetryQueueClone(int slot_idx, const mesh::Packet* packet, + uint8_t attempt_idx); + uint32_t companionRetryDelay(const mesh::Packet* packet, bool direct, uint8_t attempt_idx); + void writeOKFrame(); void writeErrFrame(uint8_t err_code); void writeDisabledFrame(); @@ -1143,17 +1168,47 @@ private: int proto_num_clients; struct AckTableEntry { - unsigned long msg_sent; - uint32_t ack; - ContactInfo* contact; + unsigned long msg_sent = 0; + uint32_t ack = 0; + ContactInfo* contact = nullptr; + uint8_t text_fingerprint[MAX_HASH_SIZE] = {}; + uint8_t retry_key[MAX_HASH_SIZE] = {}; }; #define EXPECTED_ACK_TABLE_SIZE 8 AckTableEntry expected_ack_table[EXPECTED_ACK_TABLE_SIZE]; // circular table int next_ack_idx; + AckTableEntry* findPendingTextMessage(const uint8_t text_fingerprint[MAX_HASH_SIZE]); + void clearExpectedAck(AckTableEntry& entry, bool cancel_retry); + + uint32_t _last_plain_tx_ack = 0; + uint8_t _last_plain_tx_fingerprint[MAX_HASH_SIZE] = {}; + uint8_t _last_plain_tx_retry_key[MAX_HASH_SIZE] = {}; + bool _last_plain_tx_meta_valid = false; #define ADVERT_PATH_TABLE_SIZE 16 AdvertPath advert_paths[ADVERT_PATH_TABLE_SIZE]; // circular table + // Client-side retry-until-echo state. Only one exact packet clone is retained + // in the outbound queue per active slot, matching the companion behavior. + static const uint8_t COMPANION_RETRY_SLOTS = 6; + struct CompanionRetrySlot { + mesh::Packet* queued_packet = nullptr; + uint32_t retry_at = 0; + uint32_t retry_delay = 0; + uint32_t missing_since = 0; + uint8_t retry_key[MAX_HASH_SIZE] = {}; + uint8_t attempts_sent = 0; + uint8_t max_attempts = 0; + uint8_t priority = 0; + uint8_t progress_marker = 0; + uint8_t payload_type = 0; + bool direct = false; + bool waiting_final_echo = false; + bool active = false; + }; + CompanionRetrySlot _companion_retries[COMPANION_RETRY_SLOTS]; + uint8_t _active_companion_retries = 0; + // One-shot auto-advert on boot. Recipients with auto-add ON pick up our // current pubkey, which is critical when the touch firmware regenerates // identity after a SPIFFS wipe (otherwise old contacts have stale pubkey diff --git a/src/helpers/CompanionRetryPolicy.h b/src/helpers/CompanionRetryPolicy.h new file mode 100644 index 0000000..826d120 --- /dev/null +++ b/src/helpers/CompanionRetryPolicy.h @@ -0,0 +1,64 @@ +#pragma once + +#include +#include +#include + +namespace CompanionRetryPolicy { + +// These are the companion defaults from MeshCore's keymindCascade branch. +static const uint8_t DIRECT_MAX_ATTEMPTS = 21; +static const uint8_t FLOOD_MAX_ATTEMPTS = 15; + +inline uint32_t directDelay(uint32_t packet_airtime_ms, uint8_t attempt_idx) { + return 200UL + (7UL * packet_airtime_ms) + (100UL * attempt_idx); +} + +inline uint32_t floodDelay(uint32_t max_packet_airtime_ms, + uint32_t packet_airtime_ms, + uint32_t jitter_percent) { + return max_packet_airtime_ms + (20UL * packet_airtime_ms) + + ((packet_airtime_ms * jitter_percent) / 100UL); +} + +inline bool isDirectEcho(uint8_t original_path_count, uint8_t received_path_count) { + return received_path_count < original_path_count; +} + +inline bool isFloodEcho(uint8_t original_path_count, uint8_t received_path_count) { + return received_path_count > original_path_count; +} + +inline bool keyIsSet(const uint8_t* key, size_t key_size) { + if (!key) return false; + for (size_t i = 0; i < key_size; i++) { + if (key[i] != 0) return true; + } + return false; +} + +inline bool keysEqual(const uint8_t* first, const uint8_t* second, size_t key_size) { + return first && second && memcmp(first, second, key_size) == 0; +} + +inline bool shouldReplacePendingText(uint32_t pending_ack, + const uint8_t* pending_fingerprint, + const uint8_t* new_fingerprint, + size_t fingerprint_size) { + return pending_ack != 0 + && keysEqual(pending_fingerprint, new_fingerprint, fingerprint_size); +} + +inline bool ackMatches(uint32_t expected_ack, const uint8_t received_ack[4]) { + return expected_ack != 0 && received_ack + && memcmp(received_ack, &expected_ack, sizeof(expected_ack)) == 0; +} + +// Wrap-safe for deadlines less than INT32_MAX milliseconds away. Retry delays +// are measured in seconds, so they remain comfortably inside that window. +inline uint32_t wakeDelay(uint32_t now_millis, uint32_t deadline_millis) { + const int32_t signed_delay = static_cast(deadline_millis - now_millis); + return signed_delay > 0 ? static_cast(signed_delay) : 0; +} + +} // namespace CompanionRetryPolicy diff --git a/src/ui-touch/TouchSleep.cpp b/src/ui-touch/TouchSleep.cpp index 85102c1..1a571b2 100644 --- a/src/ui-touch/TouchSleep.cpp +++ b/src/ui-touch/TouchSleep.cpp @@ -47,7 +47,7 @@ void onTransition(TransitionCb cb) { g_transition = cb; } void setEnabled(bool on) { g_enabled = on; } bool enabled() { return g_enabled; } -void loopEnd(uint32_t /*now_ms*/) { +void loopEnd(uint32_t now_ms) { const bool pass = gatePasses(); if (!pass) { if (g_asleep_regime) { g_asleep_regime = false; emitTransition(false); } // sun — resumed activity @@ -55,10 +55,23 @@ void loopEnd(uint32_t /*now_ms*/) { } if (!g_asleep_regime) { g_asleep_regime = true; emitTransition(true); } // moon — parked - // Yield the CPU to the idle task for THROTTLE_MS (replaces esp_light_sleep_start — - // see the note above). Standard FreeRTOS call: cannot starve a watchdog or hang. + // Yield the CPU to the idle task, capped by the earliest retry/UI deadline. + // This is the wake timer for these builds: unlike manual esp_light_sleep_start, + // a timed FreeRTOS block remains watchdog-safe and resumes the loop on schedule. + uint32_t park_ms = THROTTLE_MS; + if (g_hooks.nextWakeForcingDueMs) { + const uint32_t wake_due_ms = g_hooks.nextWakeForcingDueMs(now_ms); + if (wake_due_ms < park_ms) park_ms = wake_due_ms; + } + if (park_ms == 0) { + g_last_reason = WakeReason::Timer; + return; + } + const uint64_t t0 = esp_timer_get_time(); - vTaskDelay(pdMS_TO_TICKS(THROTTLE_MS)); + TickType_t park_ticks = pdMS_TO_TICKS(park_ms); + if (park_ticks == 0) park_ticks = 1; + vTaskDelay(park_ticks); g_acc_idle_us += (uint64_t)(esp_timer_get_time() - t0); g_cycle_count++; g_last_reason = WakeReason::Timer; // a throttle is a timed yield diff --git a/src/ui-touch/TouchSleep.h b/src/ui-touch/TouchSleep.h index 1360718..f93622c 100644 --- a/src/ui-touch/TouchSleep.h +++ b/src/ui-touch/TouchSleep.h @@ -20,7 +20,7 @@ struct Hooks { bool (*bleOff)(); // BLE fully off bool (*onBattery)(); // running on battery (no USB / charge source) bool (*meshIdle)(); // radio not mid-RX AND send queue empty - // ms until the soonest wake-forcing deadline (advert / clock alarm), + // ms until the soonest wake-forcing deadline (retry / advert / clock alarm), // or UINT32_MAX when nothing must wake us: uint32_t (*nextWakeForcingDueMs)(uint32_t now_ms); uint32_t (*epochNow)(); // wall-clock epoch seconds (for the event log) diff --git a/src/ui-touch/UITask.cpp b/src/ui-touch/UITask.cpp index 39fe12b..eabb778 100644 --- a/src/ui-touch/UITask.cpp +++ b/src/ui-touch/UITask.cpp @@ -15231,6 +15231,7 @@ static void actionSheetRangeTestCb(lv_event_t* e) { uint32_t ack_hash = 0; int r = the_mesh.sendMessage(c, ts, 0, "RangeTest \xe2\x80\x94 ACK?", ack_hash, est, &hash4); if (r == MSG_SEND_SENT_FLOOD || r == MSG_SEND_SENT_DIRECT) { + the_mesh.uiRegisterExpectedAck(ack_hash, c.id.pub_key); markMeshRequest(); // status-bar async spinner g_lv.task->showAlert(r == MSG_SEND_SENT_DIRECT ? TR("RangeTest sent (direct)") : TR("RangeTest sent (flood)"), 1400); @@ -37697,15 +37698,23 @@ static bool tsBleOff() { // — batteryIsCharging(batteryMvSmoothed())). static bool tsOnBattery() { return !batteryIsCharging(batteryMvSmoothed()); } // tsMeshIdle: true when the radio is NOT mid-receive (preamble→RxDone race guarded), -// AND no outbound packets are queued / no dirty contacts expiry is pending. -// hasPendingWork() checks _mgr->getOutboundTotal() + dirty_contacts_expiry. +// AND no outbound packet / retry / contact write is due now. Future retry and +// write deadlines are allowed to use the timed idle-power-saving path. // isRadioReceiving() delegates to Dispatcher::_radio->isReceiving() (RadioLibWrapper override). static bool tsMeshIdle() { return !the_mesh.hasPendingWork() && !the_mesh.isRadioReceiving(); } -// tsNextWakeForcingDueMs: advert/sig-probe is the only wake-forcing deadline today; -// clock alarms TBD. s_sig_probe_at is promoted to file scope so we can read it here. +// Bound the timed idle park by the earliest UI probe or companion retry. static uint32_t tsNextWakeForcingDueMs(uint32_t now_ms) { - if (s_sig_probe_at == 0) return UINT32_MAX; - return (uint32_t)(s_sig_probe_at > now_ms ? (s_sig_probe_at - now_ms) : 0); + uint32_t next_delay = UINT32_MAX; + if (s_sig_probe_at != 0) { + next_delay = (uint32_t)(s_sig_probe_at > now_ms ? (s_sig_probe_at - now_ms) : 0); + } + + uint32_t retry_delay = 0; + if (the_mesh.getNextCompanionRetryWakeDelay(retry_delay) + && retry_delay < next_delay) { + next_delay = retry_delay; + } + return next_delay; } static uint32_t tsEpochNow() { return (uint32_t)time(nullptr); } diff --git a/test/test_companion_retry_policy.cpp b/test/test_companion_retry_policy.cpp new file mode 100644 index 0000000..96da551 --- /dev/null +++ b/test/test_companion_retry_policy.cpp @@ -0,0 +1,55 @@ +#include + +#include "helpers/CompanionRetryPolicy.h" + +int main() { + using namespace CompanionRetryPolicy; + + static_assert(DIRECT_MAX_ATTEMPTS == 21, "direct retry count changed"); + static_assert(FLOOD_MAX_ATTEMPTS == 15, "flood retry count changed"); + + assert(directDelay(100, 0) == 900); + assert(directDelay(100, 20) == 2900); + + assert(floodDelay(300, 100, 0) == 2300); + assert(floodDelay(300, 100, 100) == 2400); + assert(floodDelay(300, 100, 200) == 2500); + + assert(isDirectEcho(5, 4)); + assert(isDirectEcho(5, 0)); + assert(!isDirectEcho(5, 5)); + assert(!isDirectEcho(5, 6)); + + assert(isFloodEcho(0, 1)); + assert(isFloodEcho(2, 5)); + assert(!isFloodEcho(2, 2)); + assert(!isFloodEcho(2, 1)); + + const uint8_t empty_key[8] = {}; + const uint8_t retry_key[8] = {1, 2, 3, 4, 5, 6, 7, 8}; + const uint8_t same_retry_key[8] = {1, 2, 3, 4, 5, 6, 7, 8}; + const uint8_t prefix_collision[8] = {1, 2, 3, 4, 5, 6, 7, 9}; + assert(!keyIsSet(empty_key, sizeof(empty_key))); + assert(keyIsSet(retry_key, sizeof(retry_key))); + assert(keysEqual(retry_key, same_retry_key, sizeof(retry_key))); + assert(!keysEqual(retry_key, prefix_collision, sizeof(retry_key))); + + assert(shouldReplacePendingText(123, retry_key, same_retry_key, + sizeof(retry_key))); + assert(!shouldReplacePendingText(123, retry_key, prefix_collision, + sizeof(retry_key))); + assert(!shouldReplacePendingText(0, retry_key, same_retry_key, + sizeof(retry_key))); + + uint32_t expected_ack = 0x12345678UL; + uint8_t received_ack[4]; + memcpy(received_ack, &expected_ack, sizeof(received_ack)); + assert(ackMatches(expected_ack, received_ack)); + assert(!ackMatches(0, empty_key)); + + assert(wakeDelay(1000, 1500) == 500); + assert(wakeDelay(1500, 1500) == 0); + assert(wakeDelay(1600, 1500) == 0); + assert(wakeDelay(0xFFFFFFF0UL, 0x00000020UL) == 48); + return 0; +}