diff --git a/docs/cli_commands.md b/docs/cli_commands.md index 5598dba550..1fca1d4823 100644 --- a/docs/cli_commands.md +++ b/docs/cli_commands.md @@ -1123,6 +1123,34 @@ region save --- +#### View the bridge frame counters +**Usage:** +- `get bridge.rxstats` +- `get bridge.txstats` +- `set bridge.stats reset` + +Counts every way a frame can be lost, so a message-count mismatch between two +bridged nodes can be attributed rather than guessed at. Compare the sending +node's TX count against the receiving node's `RX in` to separate loss on the +link from local drops. + +Common RX fields: `in` frames received, `ok` queued to the mesh, `dup` already +seen, `nopar` not a valid mesh packet, `pool` packet pool was empty (frame +retried, not lost), `qfull` lost because the main loop was not draining fast +enough, `hwm` deepest the receive queue has been. + +**ESP-NOW** adds `bad` (wrong magic/checksum, usually another network's secret) +on RX, and on TX reports `ok` frames the radio confirmed, `dup` already bridged, +`big` too large for an ESP-NOW frame, `rty` retransmissions, `ref` radio refused +the frame, `tmo` send callback never arrived, `fail` abandoned after every +attempt, `qfull` transmit queue full, `hwm` transmit queue depth. + +**RS232** adds `crc` (checksum failures), `len` (bad length field) and `noise` +(bytes skipped while hunting for a frame header) on RX, which together indicate +a marginal cable or a baud-rate mismatch. Its TX side reports `dup` and `big`. + +--- + #### View the bootloader version (nRF52 only) **Usage:** `get bootloader.ver` diff --git a/examples/simple_repeater/MyMesh.h b/examples/simple_repeater/MyMesh.h index 0b2e7491b7..0a1d9ee674 100644 --- a/examples/simple_repeater/MyMesh.h +++ b/examples/simple_repeater/MyMesh.h @@ -249,6 +249,23 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks { } #endif +#if defined(WITH_BRIDGE) + bool getBridgeRxStats(char* dest, size_t dest_size) override { + bridge.getRxStats(dest, dest_size); + return true; + } + + bool getBridgeTxStats(char* dest, size_t dest_size) override { + bridge.getTxStats(dest, dest_size); + return true; + } + + bool resetBridgeStats() override { + bridge.resetStats(); + return true; + } +#endif + // To check if there is pending work bool hasPendingWork() const; diff --git a/platformio.ini b/platformio.ini index cdd67d2770..32140c9699 100644 --- a/platformio.ini +++ b/platformio.ini @@ -51,6 +51,10 @@ build_src_filter = + + + + + + + + + + + + ; ----------------- ESP32 --------------------- @@ -163,11 +167,17 @@ build_flags = -std=c++17 -I src -I test/mocks test_build_src = yes -test_filter = test_utils +test_filter = + test_utils + test_bridge build_src_filter = -<*> +<../src/Utils.cpp> +<../src/Packet.cpp> + +<../src/helpers/bridges/BridgeCodec.cpp> + +<../src/helpers/bridges/BridgeSerialFramer.cpp> + +<../src/helpers/bridges/BridgeFrameQueue.cpp> + +<../src/helpers/bridges/BridgeTxQueue.cpp> lib_deps = google/googletest @ 1.17.0 diff --git a/src/helpers/CommonCLI.cpp b/src/helpers/CommonCLI.cpp index c95e3e34b0..039b2fa4fa 100644 --- a/src/helpers/CommonCLI.cpp +++ b/src/helpers/CommonCLI.cpp @@ -297,7 +297,7 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, char* command, char* re StrHelper::strncpy(_prefs->password, &command[9], sizeof(_prefs->password)); savePrefs(); sprintf(reply, "password now: "); - StrHelper::strncpy(&reply[14], _prefs->password, 160-15); // echo back just to let admin know for sure!! + StrHelper::strncpy(&reply[14], _prefs->password, MAX_REPLY_LEN-15); // echo back just to let admin know for sure!! } else if (memcmp(command, "clear stats", 11) == 0) { _callbacks->clearStats(); strcpy(reply, "(OK - stats reset)"); @@ -750,6 +750,9 @@ void CommonCLI::handleSetCmd(uint32_t sender_timestamp, char* command, char* rep _prefs->bridge_pkt_src = memcmp(&config[14], "rx", 2) == 0; savePrefs(); strcpy(reply, "OK"); + } else if (memcmp(config, "bridge.stats reset", 18) == 0) { + // zero the counters so a measurement can start from a known point + strcpy(reply, _callbacks->resetBridgeStats() ? "OK" : "Bridge stats not supported"); #endif #ifdef WITH_RS232_BRIDGE } else if (memcmp(config, "bridge.baud ", 12) == 0) { @@ -795,7 +798,7 @@ void CommonCLI::handleSetCmd(uint32_t sender_timestamp, char* command, char* rep }; } else { strcpy(reply, "unknown config: "); - StrHelper::strncpy(&reply[16], config, 160-17); + StrHelper::strncpy(&reply[16], config, MAX_REPLY_LEN-17); } } @@ -910,6 +913,18 @@ void CommonCLI::handleGetCmd(uint32_t sender_timestamp, char* command, char* rep sprintf(reply, "> %d", (uint32_t)_prefs->bridge_delay); } else if (memcmp(config, "bridge.source", 13) == 0) { sprintf(reply, "> %s", _prefs->bridge_pkt_src ? "logRx" : "logTx"); + } else if (memcmp(config, "bridge.rxstats", 14) == 0) { + // Every way the bridge can lose a frame, so a mismatch between two bridged + // nodes can be attributed instead of guessed at. + strcpy(reply, "> "); + if (!_callbacks->getBridgeRxStats(&reply[2], MAX_REPLY_LEN - 2)) { + strcpy(reply, "Bridge stats not supported"); + } + } else if (memcmp(config, "bridge.txstats", 14) == 0) { + strcpy(reply, "> "); + if (!_callbacks->getBridgeTxStats(&reply[2], MAX_REPLY_LEN - 2)) { + strcpy(reply, "Bridge stats not supported"); + } #endif #ifdef WITH_RS232_BRIDGE } else if (memcmp(config, "bridge.baud", 11) == 0) { diff --git a/src/helpers/CommonCLI.h b/src/helpers/CommonCLI.h index f3abcf4772..d52c811665 100644 --- a/src/helpers/CommonCLI.h +++ b/src/helpers/CommonCLI.h @@ -111,11 +111,36 @@ class CommonCLICallbacks { // no op by default }; + /** + * Writes the bridge's receive-side counters into dest. + * @return false if this build has no bridge that reports statistics + */ + virtual bool getBridgeRxStats(char* dest, size_t dest_size) { + return false; + }; + + /** Writes the bridge's transmit-side counters into dest. */ + virtual bool getBridgeTxStats(char* dest, size_t dest_size) { + return false; + }; + + /** Zeroes the bridge's counters. @return false if unsupported */ + virtual bool resetBridgeStats() { + return false; + }; + virtual bool setRxBoostedGain(bool enable) { return false; // CommonCLI reports unsupported if not overridden by wrapper }; }; +/** + * Size of the reply buffer every handleCommand() caller provides. The serial + * console uses char[160] and remote admin carves 161 bytes out of a 166 byte + * frame, so 160 is what a reply can safely assume. + */ +#define MAX_REPLY_LEN 160 + class CommonCLI { mesh::RTCClock* _rtc; NodePrefs* _prefs; diff --git a/src/helpers/bridges/BridgeBase.cpp b/src/helpers/bridges/BridgeBase.cpp index 8093d3cb5b..3f21cb8a98 100644 --- a/src/helpers/bridges/BridgeBase.cpp +++ b/src/helpers/bridges/BridgeBase.cpp @@ -15,20 +15,51 @@ const char *BridgeBase::getLogDateTime() { return tmp; } -uint16_t BridgeBase::fletcher16(const uint8_t *data, size_t len) { - uint8_t sum1 = 0, sum2 = 0; +void BridgeBase::resetBaseStats() { + _rx_delivered = 0; + _rx_duplicates = 0; + _rx_no_packet = 0; + _rx_unparsed = 0; + _tx_duplicates = 0; + _tx_oversized = 0; +} - for (size_t i = 0; i < len; i++) { - sum1 = (sum1 + data[i]) % 255; - sum2 = (sum2 + sum1) % 255; - } +void BridgeBase::drainRxFrames(BridgeFrameQueue &queue, uint8_t max_frames, uint8_t *scratch, + size_t scratch_cap) { + for (uint8_t drained = 0; drained < max_frames && !queue.isEmpty(); drained++) { + // Take a pool slot first: if the pool is empty the frame stays queued for a + // later loop, costing latency instead of a message. + mesh::Packet *pkt = _mgr->allocNew(); + if (pkt == nullptr) { + _rx_no_packet++; + break; + } - return (sum2 << 8) | sum1; -} + const size_t frame_len = queue.pop(scratch, scratch_cap); + if (frame_len == 0) { + _mgr->free(pkt); + break; + } -bool BridgeBase::validateChecksum(const uint8_t *data, size_t len, uint16_t received_checksum) { - uint16_t calculated_checksum = fletcher16(data, len); - return received_checksum == calculated_checksum; + size_t blob_len = 0; + const uint8_t *blob = unwrapFrame(scratch, frame_len, blob_len); + if (blob == nullptr) { + _mgr->free(pkt); // the override counted why + continue; + } + + BRIDGE_DEBUG_PRINTLN("RX, len=%d\n", (int)blob_len); + + // readFrom() takes a uint8_t length, and a mesh packet can never be longer + // than MAX_TRANS_UNIT + 1 anyway. + if (blob_len > 0 && blob_len <= 255 && pkt->readFrom(blob, (uint8_t)blob_len)) { + onPacketReceived(pkt); // takes ownership + } else { + BRIDGE_DEBUG_PRINTLN("RX failed to parse packet\n"); + _rx_unparsed++; + _mgr->free(pkt); + } + } } void BridgeBase::handleReceivedPacket(mesh::Packet *packet) { @@ -43,7 +74,9 @@ void BridgeBase::handleReceivedPacket(mesh::Packet *packet) { _seen_packets.markSeen(packet); // bridge_delay provides a buffer to prevent immediate processing conflicts in the mesh network. _mgr->queueInbound(packet, millis() + _prefs->bridge_delay); + _rx_delivered++; } else { _mgr->free(packet); + _rx_duplicates++; } } diff --git a/src/helpers/bridges/BridgeBase.h b/src/helpers/bridges/BridgeBase.h index 8bbe646678..b55cb471b9 100644 --- a/src/helpers/bridges/BridgeBase.h +++ b/src/helpers/bridges/BridgeBase.h @@ -3,6 +3,7 @@ #include "helpers/AbstractBridge.h" #include "helpers/CommonCLI.h" #include "helpers/SimpleMeshTables.h" +#include "helpers/bridges/BridgeFrameQueue.h" #include @@ -10,13 +11,16 @@ * @brief Base class implementing common bridge functionality * * This class provides common functionality used by different bridge implementations - * like packet tracking, checksum calculation, timestamping, and duplicate detection. + * like packet tracking, timestamping, and duplicate detection. * * Features: - * - Fletcher-16 checksum calculation for data integrity * - Packet duplicate detection using SimpleMeshTables * - Common timestamp formatting for debug logging * - Shared packet management and queuing logic + * - Counters shared by every bridge, behind `get bridge.rxstats`/`bridge.txstats` + * + * The wire framing itself lives in BridgeCodec/BridgeSerialFramer, which stay + * free of the Arduino and mesh headers so they can be unit tested natively. */ class BridgeBase : public AbstractBridge { public: @@ -29,26 +33,6 @@ class BridgeBase : public AbstractBridge { */ bool isRunning() const override; - /** - * @brief Common magic number used by all bridge implementations for packet identification - * - * This magic number is placed at the beginning of bridge packets to identify - * them as mesh bridge packets and provide frame synchronization. - */ - static constexpr uint16_t BRIDGE_PACKET_MAGIC = 0xC03E; - - /** - * @brief Common field sizes used by bridge implementations - * - * These constants define the size of common packet fields used across bridges. - * BRIDGE_MAGIC_SIZE is used by all bridges for packet identification. - * BRIDGE_LENGTH_SIZE is used by bridges that need explicit length fields (like RS232). - * BRIDGE_CHECKSUM_SIZE is used by all bridges for Fletcher-16 checksums. - */ - static constexpr uint16_t BRIDGE_MAGIC_SIZE = sizeof(BRIDGE_PACKET_MAGIC); - static constexpr uint16_t BRIDGE_LENGTH_SIZE = sizeof(uint16_t); - static constexpr uint16_t BRIDGE_CHECKSUM_SIZE = sizeof(uint16_t); - protected: /** Tracks bridge state */ bool _initialized = false; @@ -65,6 +49,24 @@ class BridgeBase : public AbstractBridge { /** Tracks seen packets to prevent loops in broadcast communications */ SimpleMeshTables _seen_packets; + /** Packets handed to the mesh inbound queue */ + uint32_t _rx_delivered = 0; + + /** Packets dropped on arrival because this node had already seen them */ + uint32_t _rx_duplicates = 0; + + /** Frames left queued because the mesh packet pool was empty */ + uint32_t _rx_no_packet = 0; + + /** Frames that arrived intact but were not a valid mesh packet */ + uint32_t _rx_unparsed = 0; + + /** Packets not bridged because this node had already sent them */ + uint32_t _tx_duplicates = 0; + + /** Packets too large to frame for this transport */ + uint32_t _tx_oversized = 0; + /** * @brief Constructs a BridgeBase instance * @@ -84,27 +86,44 @@ class BridgeBase : public AbstractBridge { */ const char *getLogDateTime(); + /** Zeroes the counters owned by this class. Call from the bridge's resetStats(). */ + void resetBaseStats(); + /** - * @brief Calculate Fletcher-16 checksum + * @brief Strip the transport's framing off a received frame. * - * Based on: https://en.wikipedia.org/wiki/Fletcher%27s_checksum - * Used to verify data integrity of received packets + * Called by drainRxFrames() once per frame. The default treats the frame as a + * bare mesh packet blob, which is what a transport that carries no framing of + * its own (RS232, whose framer has already unwrapped it) needs. * - * @param data Pointer to data to calculate checksum for - * @param len Length of data in bytes - * @return Calculated Fletcher-16 checksum + * @param frame the frame as popped from the receive queue + * @param frame_len length of @p frame + * @param blob_len set to the length of the returned blob + * @return pointer to the mesh packet blob, or nullptr to drop the frame. An + * override returning nullptr is expected to count its own reason. */ - static uint16_t fletcher16(const uint8_t *data, size_t len); + virtual const uint8_t *unwrapFrame(const uint8_t *frame, size_t frame_len, size_t &blob_len) { + blob_len = frame_len; + return frame; + } /** - * @brief Validate received checksum against calculated checksum + * @brief Turn queued frames into mesh packets, bounded per call. + * + * Shared by every bridge because the ownership rules are subtle: a pool slot is + * taken *before* the frame is popped, so an exhausted pool leaves the frame + * queued for a later loop instead of destroying it, and every path that does + * not hand the packet to the mesh frees it exactly once. * - * @param data Pointer to data to validate - * @param len Length of data in bytes - * @param received_checksum Checksum received with data - * @return true if checksum is valid, false otherwise + * @param queue frames waiting to be delivered + * @param max_frames frames to convert before returning, so one bridge cannot + * monopolise the loop + * @param scratch buffer a frame is popped into; must hold the largest frame + * the queue accepts + * @param scratch_cap capacity of @p scratch */ - bool validateChecksum(const uint8_t *data, size_t len, uint16_t received_checksum); + void drainRxFrames(BridgeFrameQueue &queue, uint8_t max_frames, uint8_t *scratch, + size_t scratch_cap); /** * @brief Common packet handling for received packets diff --git a/src/helpers/bridges/BridgeCodec.cpp b/src/helpers/bridges/BridgeCodec.cpp new file mode 100644 index 0000000000..c0b7c0a2fb --- /dev/null +++ b/src/helpers/bridges/BridgeCodec.cpp @@ -0,0 +1,71 @@ +#include "BridgeCodec.h" + +#include + +uint16_t BridgeCodec::fletcher16(const uint8_t *data, size_t len) { + uint8_t sum1 = 0, sum2 = 0; + + for (size_t i = 0; i < len; i++) { + sum1 = (sum1 + data[i]) % 255; + sum2 = (sum2 + sum1) % 255; + } + + return (uint16_t)((sum2 << 8) | sum1); +} + +void BridgeCodec::xorCrypt(uint8_t *data, size_t len, const char *secret, size_t offset) { + if (secret == nullptr) return; + + const size_t key_len = strlen(secret); + if (key_len == 0) return; // blank secret means "no encryption", never a divide by zero + + // Walk the key rather than reducing every index: key_len is a runtime value, so + // a modulo per byte is a real division on the Cortex-M targets. + size_t k = offset % key_len; + for (size_t i = 0; i < len; i++) { + data[i] ^= (uint8_t)secret[k]; + if (++k == key_len) k = 0; + } +} + +int BridgeCodec::encode(const uint8_t *payload, size_t payload_len, const char *secret, uint8_t *out, + size_t out_cap) { + if (payload == nullptr || payload_len == 0) return -1; + + const size_t frame_len = FRAME_OVERHEAD + payload_len; + if (frame_len > out_cap) return -1; // refuse rather than overrun the caller's buffer + + out[0] = (uint8_t)((BRIDGE_PACKET_MAGIC >> 8) & 0xFF); + out[1] = (uint8_t)(BRIDGE_PACKET_MAGIC & 0xFF); + + const uint16_t checksum = fletcher16(payload, payload_len); + out[2] = (uint8_t)((checksum >> 8) & 0xFF); + out[3] = (uint8_t)(checksum & 0xFF); + + memcpy(out + FRAME_OVERHEAD, payload, payload_len); + xorCrypt(out + BRIDGE_MAGIC_SIZE, BRIDGE_CHECKSUM_SIZE + payload_len, secret); + + return (int)frame_len; +} + +int BridgeCodec::decode(const uint8_t *frame, size_t frame_len, const char *secret, uint8_t *out, + size_t out_cap) { + if (frame == nullptr || frame_len <= FRAME_OVERHEAD) return -1; // truncated, or carries no payload + + const uint16_t magic = (uint16_t)((frame[0] << 8) | frame[1]); + if (magic != BRIDGE_PACKET_MAGIC) return -1; + + const size_t payload_len = frame_len - FRAME_OVERHEAD; + if (payload_len > out_cap) return -1; + + uint8_t checksum_bytes[BRIDGE_CHECKSUM_SIZE] = { frame[2], frame[3] }; + xorCrypt(checksum_bytes, BRIDGE_CHECKSUM_SIZE, secret); + const uint16_t received_checksum = (uint16_t)((checksum_bytes[0] << 8) | checksum_bytes[1]); + + memcpy(out, frame + FRAME_OVERHEAD, payload_len); + xorCrypt(out, payload_len, secret, BRIDGE_CHECKSUM_SIZE); + + if (fletcher16(out, payload_len) != received_checksum) return -1; + + return (int)payload_len; +} diff --git a/src/helpers/bridges/BridgeCodec.h b/src/helpers/bridges/BridgeCodec.h new file mode 100644 index 0000000000..1f7448404e --- /dev/null +++ b/src/helpers/bridges/BridgeCodec.h @@ -0,0 +1,78 @@ +#pragma once + +#include +#include + +/** + * @brief Wire framing shared by every bridge implementation. + * + * Kept free of Arduino/ESP-IDF dependencies so it can be unit tested natively. + * + * Frame layout produced by encode(): + * [2 bytes] magic header (BRIDGE_PACKET_MAGIC, big endian, sent in clear) + * [2 bytes] Fletcher-16 of the payload (big endian, encrypted) + * [n bytes] payload (encrypted) + * + * The checksum is calculated over the plaintext payload and then encrypted along + * with it, so a frame from a network using a different secret decrypts to garbage + * and fails validation. That is what provides network isolation. + */ +class BridgeCodec { +public: + /** Magic number identifying a bridge frame and providing frame sync. */ + static constexpr uint16_t BRIDGE_PACKET_MAGIC = 0xC03E; + + static constexpr uint16_t BRIDGE_MAGIC_SIZE = sizeof(uint16_t); + static constexpr uint16_t BRIDGE_LENGTH_SIZE = sizeof(uint16_t); + static constexpr uint16_t BRIDGE_CHECKSUM_SIZE = sizeof(uint16_t); + + /** Bytes of framing overhead that encode() adds around the payload. */ + static constexpr size_t FRAME_OVERHEAD = BRIDGE_MAGIC_SIZE + BRIDGE_CHECKSUM_SIZE; + + /** + * @brief Fletcher-16 checksum. + * + * Based on https://en.wikipedia.org/wiki/Fletcher%27s_checksum + */ + static uint16_t fletcher16(const uint8_t *data, size_t len); + + /** + * @brief XOR a buffer with a repeating secret, in place. + * + * @p offset says how far into the encrypted region @p data starts, which lets + * decode() decrypt the checksum and the payload separately without staging the + * whole frame in a temporary buffer. + * + * Does nothing when @p secret is null or empty, so a node configured with a + * blank secret still interoperates (unencrypted) instead of dividing by zero. + */ + static void xorCrypt(uint8_t *data, size_t len, const char *secret, size_t offset = 0); + + /** + * @brief Frame and encrypt a mesh packet blob. + * + * @param payload the mesh packet as produced by Packet::writeTo() + * @param payload_len length of @p payload + * @param secret shared network secret (may be null/empty) + * @param out destination buffer + * @param out_cap capacity of @p out + * @return encoded frame length, or -1 if the frame would not fit in @p out_cap + * (in which case @p out is left untouched) + */ + static int encode(const uint8_t *payload, size_t payload_len, const char *secret, uint8_t *out, + size_t out_cap); + + /** + * @brief Validate, decrypt and unwrap a frame produced by encode(). + * + * @param frame the received frame, starting at the magic header + * @param frame_len length of @p frame + * @param secret shared network secret (may be null/empty) + * @param out destination for the decrypted payload + * @param out_cap capacity of @p out + * @return payload length written to @p out, or -1 if the frame is truncated, + * has the wrong magic, fails its checksum, or will not fit @p out_cap + */ + static int decode(const uint8_t *frame, size_t frame_len, const char *secret, uint8_t *out, + size_t out_cap); +}; diff --git a/src/helpers/bridges/BridgeFrameQueue.cpp b/src/helpers/bridges/BridgeFrameQueue.cpp new file mode 100644 index 0000000000..229f958221 --- /dev/null +++ b/src/helpers/bridges/BridgeFrameQueue.cpp @@ -0,0 +1,112 @@ +#include "BridgeFrameQueue.h" + +#include + +namespace { + +/** + * Head and tail run over [0, 2*capacity), not the whole uint32_t range: reducing + * a free-running counter modulo a non-power-of-two capacity would alias two + * slots onto each other at the 2^32 wrap. + */ +inline uint32_t ringSpan(uint8_t capacity) { + return 2u * (uint32_t)capacity; +} + +inline uint32_t ringCount(uint32_t head, uint32_t tail, uint8_t capacity) { + const uint32_t span = ringSpan(capacity); + return (head + span - tail) % span; +} + +} // namespace + +BridgeFrameQueue::BridgeFrameQueue(uint8_t capacity, uint16_t max_frame_len) + : _capacity(capacity < 1 ? 1 : capacity), _max_frame_len(max_frame_len), _slots(nullptr), + _lengths(nullptr), _head(0), _tail(0), _pushed(0), _dropped_full(0), + _dropped_oversize_in(0), _high_water(0), _popped(0), _dropped_oversize_out(0) { + _slots = new uint8_t[(size_t)_capacity * (size_t)_max_frame_len]; + _lengths = new uint16_t[_capacity]; +} + +BridgeFrameQueue::~BridgeFrameQueue() { + delete[] _slots; + delete[] _lengths; +} + +bool BridgeFrameQueue::push(const uint8_t *frame, size_t len) { + if (frame == nullptr || len == 0 || len > _max_frame_len) { + _dropped_oversize_in.fetch_add(1, std::memory_order_relaxed); + return false; + } + + const uint32_t head = _head.load(std::memory_order_relaxed); + const uint32_t tail = _tail.load(std::memory_order_acquire); + + const uint32_t depth = ringCount(head, tail, _capacity); + if (depth >= _capacity) { + _dropped_full.fetch_add(1, std::memory_order_relaxed); + return false; + } + + const uint32_t slot = head % _capacity; + memcpy(_slots + (size_t)slot * (size_t)_max_frame_len, frame, len); + _lengths[slot] = (uint16_t)len; + + // release: the frame bytes must be visible before the consumer sees the new head + _head.store((head + 1) % ringSpan(_capacity), std::memory_order_release); + + _pushed.fetch_add(1, std::memory_order_relaxed); + if (depth + 1 > _high_water.load(std::memory_order_relaxed)) { + _high_water.store((uint8_t)(depth + 1), std::memory_order_relaxed); + } + return true; +} + +size_t BridgeFrameQueue::pop(uint8_t *out, size_t out_cap) { + const uint32_t tail = _tail.load(std::memory_order_relaxed); + const uint32_t head = _head.load(std::memory_order_acquire); + + if (head == tail) return 0; + + const uint32_t slot = tail % _capacity; + const uint16_t len = _lengths[slot]; + + size_t written = 0; + if (out == nullptr || len > out_cap) { + // Drop it: leaving it at the head would stall every frame behind it for good. + _dropped_oversize_out++; + } else { + memcpy(out, _slots + (size_t)slot * (size_t)_max_frame_len, len); + written = len; + _popped++; + } + + // release: the slot is only reusable once we are done reading it + _tail.store((tail + 1) % ringSpan(_capacity), std::memory_order_release); + + return written; +} + +void BridgeFrameQueue::clear() { + // release: matches pop(), so the producer only reuses slots we are done reading + _tail.store(_head.load(std::memory_order_acquire), std::memory_order_release); +} + +bool BridgeFrameQueue::isEmpty() const { + return _head.load(std::memory_order_acquire) == _tail.load(std::memory_order_acquire); +} + +uint8_t BridgeFrameQueue::count() const { + const uint32_t head = _head.load(std::memory_order_acquire); + const uint32_t tail = _tail.load(std::memory_order_acquire); + return (uint8_t)ringCount(head, tail, _capacity); +} + +void BridgeFrameQueue::resetStats() { + _pushed.store(0, std::memory_order_relaxed); + _dropped_full.store(0, std::memory_order_relaxed); + _dropped_oversize_in.store(0, std::memory_order_relaxed); + _high_water.store(0, std::memory_order_relaxed); + _popped = 0; + _dropped_oversize_out = 0; +} diff --git a/src/helpers/bridges/BridgeFrameQueue.h b/src/helpers/bridges/BridgeFrameQueue.h new file mode 100644 index 0000000000..a772e4ce26 --- /dev/null +++ b/src/helpers/bridges/BridgeFrameQueue.h @@ -0,0 +1,102 @@ +#pragma once + +#include +#include +#include + +/** + * @brief Fixed-capacity FIFO of raw bridge frames. + * + * Lets a bridge accept a frame in a driver-task context (the ESP-NOW receive + * callback runs on the Wi-Fi task) without touching the mesh packet pool, the + * inbound queue or the seen-packet table, none of which are synchronised. + * + * Safe for one producer and one consumer running concurrently: the producer only + * advances _head, the consumer only advances _tail, and each index is published + * with a release store and read with an acquire load. NOT safe for multiple + * producers or consumers. + * + * Frames dropped for want of room are counted, so loss is visible over the CLI. + */ +class BridgeFrameQueue { +public: + /** + * @param capacity maximum number of frames held at once + * @param max_frame_len maximum size of a single frame, in bytes + */ + BridgeFrameQueue(uint8_t capacity, uint16_t max_frame_len); + ~BridgeFrameQueue(); + + BridgeFrameQueue(const BridgeFrameQueue &) = delete; + BridgeFrameQueue &operator=(const BridgeFrameQueue &) = delete; + + /** + * @brief Append a frame. Producer side. + * + * @return true if stored; false if the frame was empty, longer than + * max_frame_len, or the queue was full. Every false bumps a counter. + */ + bool push(const uint8_t *frame, size_t len); + + /** + * @brief Remove the oldest frame. Consumer side. + * + * If the frame does not fit @p out_cap it is discarded and counted, rather + * than left at the head where it would wedge the queue permanently. + * + * @return bytes written to @p out, or 0 if the queue was empty. + */ + size_t pop(uint8_t *out, size_t out_cap); + + /** + * @brief Discard every queued frame. Consumer side. + * + * Releases all slots at once, so teardown does not need a throwaway buffer to + * pop into and the dropped-frame counters are left alone. + */ + void clear(); + + bool isEmpty() const; + uint8_t count() const; + uint8_t capacity() const { return _capacity; } + uint16_t maxFrameLen() const { return _max_frame_len; } + + uint32_t getPushed() const { return _pushed.load(std::memory_order_relaxed); } + uint32_t getPopped() const { return _popped; } + /** Frames lost because the queue was full — the consumer is not keeping up. */ + uint32_t getDroppedFull() const { return _dropped_full.load(std::memory_order_relaxed); } + /** + * Frames rejected on the way in because they were empty or longer than + * max_frame_len, plus frames discarded on the way out because they did not fit + * the consumer's buffer. + */ + uint32_t getDroppedOversize() const { + return _dropped_oversize_in.load(std::memory_order_relaxed) + _dropped_oversize_out; + } + /** Deepest the queue has ever been; shows how close to full it runs. */ + uint8_t getHighWaterMark() const { return _high_water.load(std::memory_order_relaxed); } + + void resetStats(); + +private: + uint8_t _capacity; + uint16_t _max_frame_len; + uint8_t *_slots; ///< _capacity contiguous blocks of _max_frame_len bytes + uint16_t *_lengths; ///< payload length held in each slot + + std::atomic _head; ///< producer-owned, total frames ever stored + std::atomic _tail; ///< consumer-owned, total frames ever removed + + // Producer-side counters. resetStats() and the getters run on the consumer, so + // these have two writers and are atomic: a plain read-modify-write here could + // lose a reset issued from the CLI, or report a torn value. + std::atomic _pushed; + std::atomic _dropped_full; + std::atomic _dropped_oversize_in; + std::atomic _high_water; + + // Consumer-side counters. pop(), resetStats() and the getters all run on the + // consumer, so these have a single writer and need no synchronisation. + uint32_t _popped; + uint32_t _dropped_oversize_out; +}; diff --git a/src/helpers/bridges/BridgeSerialFramer.cpp b/src/helpers/bridges/BridgeSerialFramer.cpp new file mode 100644 index 0000000000..da36234c5a --- /dev/null +++ b/src/helpers/bridges/BridgeSerialFramer.cpp @@ -0,0 +1,121 @@ +#include "BridgeSerialFramer.h" + +#include + +namespace { + +constexpr uint8_t MAGIC_HI = (BridgeCodec::BRIDGE_PACKET_MAGIC >> 8) & 0xFF; +constexpr uint8_t MAGIC_LO = BridgeCodec::BRIDGE_PACKET_MAGIC & 0xFF; + +} // namespace + +BridgeSerialFramer::BridgeSerialFramer(uint16_t max_payload_len) + : _max_payload_len(max_payload_len), _payload(nullptr), _state(STATE_MAGIC_HI), + _expected_len(0), _payload_pos(0), _checksum(0), _frames_decoded(0), _checksum_errors(0), + _length_errors(0), _resync_bytes(0) { + _payload = new uint8_t[max_payload_len]; +} + +BridgeSerialFramer::~BridgeSerialFramer() { + delete[] _payload; +} + +uint16_t BridgeSerialFramer::offer(uint8_t b) { + switch (_state) { + case STATE_MAGIC_HI: + if (b == MAGIC_HI) { + _state = STATE_MAGIC_LO; + } else { + _resync_bytes++; + } + return 0; + + case STATE_MAGIC_LO: + if (b == MAGIC_LO) { + _state = STATE_LEN_HI; + } else if (b == MAGIC_HI) { + // The previous byte was noise, but this one could still start the header. + _resync_bytes++; + } else { + _resync_bytes += 2; // the earlier high byte and this one were both noise + _state = STATE_MAGIC_HI; + } + return 0; + + case STATE_LEN_HI: + _expected_len = (uint16_t)(b << 8); + _state = STATE_LEN_LO; + return 0; + + case STATE_LEN_LO: + _expected_len |= b; + if (_expected_len == 0 || _expected_len > _max_payload_len) { + // Never a valid mesh packet, and would run off the buffer. + _length_errors++; + _state = STATE_MAGIC_HI; + return 0; + } + _payload_pos = 0; + _state = STATE_PAYLOAD; + return 0; + + case STATE_PAYLOAD: + _payload[_payload_pos++] = b; + if (_payload_pos == _expected_len) { + _state = STATE_CHECKSUM_HI; + } + return 0; + + case STATE_CHECKSUM_HI: + _checksum = (uint16_t)(b << 8); + _state = STATE_CHECKSUM_LO; + return 0; + + case STATE_CHECKSUM_LO: + _checksum |= b; + _state = STATE_MAGIC_HI; + if (BridgeCodec::fletcher16(_payload, _expected_len) == _checksum) { + _frames_decoded++; + return _expected_len; + } + _checksum_errors++; + return 0; + } + + return 0; +} + +void BridgeSerialFramer::reset() { + _state = STATE_MAGIC_HI; + _expected_len = 0; + _payload_pos = 0; +} + +void BridgeSerialFramer::resetStats() { + _frames_decoded = 0; + _checksum_errors = 0; + _length_errors = 0; + _resync_bytes = 0; +} + +int BridgeSerialFramer::encode(const uint8_t *payload, size_t payload_len, uint8_t *out, + size_t out_cap) { + if (payload == nullptr || payload_len == 0) return -1; + + const size_t frame_len = FRAME_OVERHEAD + payload_len; + if (frame_len > out_cap) return -1; // refuse rather than overrun the caller's buffer + + out[0] = MAGIC_HI; + out[1] = MAGIC_LO; + out[2] = (uint8_t)((payload_len >> 8) & 0xFF); + out[3] = (uint8_t)(payload_len & 0xFF); + + memcpy(out + BridgeCodec::BRIDGE_MAGIC_SIZE + BridgeCodec::BRIDGE_LENGTH_SIZE, payload, + payload_len); + + const uint16_t checksum = BridgeCodec::fletcher16(payload, payload_len); + out[4 + payload_len] = (uint8_t)((checksum >> 8) & 0xFF); + out[5 + payload_len] = (uint8_t)(checksum & 0xFF); + + return (int)frame_len; +} diff --git a/src/helpers/bridges/BridgeSerialFramer.h b/src/helpers/bridges/BridgeSerialFramer.h new file mode 100644 index 0000000000..b2421173cf --- /dev/null +++ b/src/helpers/bridges/BridgeSerialFramer.h @@ -0,0 +1,92 @@ +#pragma once + +#include "BridgeCodec.h" + +#include +#include + +/** + * @brief Byte-at-a-time decoder for the length-prefixed serial bridge framing. + * + * Wire layout: + * [2 bytes] magic header (BridgeCodec::BRIDGE_PACKET_MAGIC, big endian) + * [2 bytes] payload length, big endian + * [n bytes] payload (a mesh packet blob) + * [2 bytes] Fletcher-16 of the payload, big endian + * + * A serial line has no packet boundaries, so the decoder resynchronises from the + * byte after any rejection and counts why, which makes a link that is quietly + * corrupting frames visible. No Arduino dependencies, so it unit tests natively. + */ +class BridgeSerialFramer { +public: + /** Bytes of framing around the payload: magic + length + checksum. */ + static constexpr uint16_t FRAME_OVERHEAD = + BridgeCodec::BRIDGE_MAGIC_SIZE + BridgeCodec::BRIDGE_LENGTH_SIZE + + BridgeCodec::BRIDGE_CHECKSUM_SIZE; + + /** + * @brief Wrap a payload in the framing this class decodes. + * + * @return framed length, or -1 if it would not fit @p out_cap (in which case + * @p out is left untouched) + */ + static int encode(const uint8_t *payload, size_t payload_len, uint8_t *out, size_t out_cap); + + /** @param max_payload_len largest payload accepted; longer length fields are rejected */ + explicit BridgeSerialFramer(uint16_t max_payload_len); + ~BridgeSerialFramer(); + + BridgeSerialFramer(const BridgeSerialFramer &) = delete; + BridgeSerialFramer &operator=(const BridgeSerialFramer &) = delete; + + /** + * @brief Feed one byte from the link. + * + * @return the payload length now available from payload(), or 0 if the frame + * is still incomplete or was rejected. The returned payload stays + * valid until the next offer() call. + */ + uint16_t offer(uint8_t b); + + /** Payload of the frame most recently completed by offer(). */ + const uint8_t *payload() const { return _payload; } + + /** Abandon any partly received frame and hunt for a fresh magic header. */ + void reset(); + + /** Complete frames that passed their checksum. */ + uint32_t getFramesDecoded() const { return _frames_decoded; } + /** Frames discarded because the payload did not match its checksum. */ + uint32_t getChecksumErrors() const { return _checksum_errors; } + /** Frames discarded because the length field was zero or too large. */ + uint32_t getLengthErrors() const { return _length_errors; } + /** Bytes thrown away while hunting for a magic header, i.e. line noise. */ + uint32_t getResyncBytes() const { return _resync_bytes; } + + void resetStats(); + +private: + enum State : uint8_t { + STATE_MAGIC_HI, + STATE_MAGIC_LO, + STATE_LEN_HI, + STATE_LEN_LO, + STATE_PAYLOAD, + STATE_CHECKSUM_HI, + STATE_CHECKSUM_LO, + }; + + uint16_t _max_payload_len; + uint8_t *_payload; + + State _state; + uint16_t _expected_len; + uint16_t _payload_pos; + uint16_t _checksum; + + uint32_t _frames_decoded; + uint32_t _checksum_errors; + uint32_t _length_errors; + uint32_t _resync_bytes; +}; diff --git a/src/helpers/bridges/BridgeTxQueue.cpp b/src/helpers/bridges/BridgeTxQueue.cpp new file mode 100644 index 0000000000..e0d4252de0 --- /dev/null +++ b/src/helpers/bridges/BridgeTxQueue.cpp @@ -0,0 +1,133 @@ +#include "BridgeTxQueue.h" + +namespace { + +/** Rollover-safe "has `deadline` been reached by `now`?". */ +inline bool hasElapsed(uint32_t now, uint32_t deadline) { + return (int32_t)(now - deadline) >= 0; +} + +} // namespace + +BridgeTxQueue::BridgeTxQueue(uint8_t capacity, uint16_t max_frame_len, uint8_t max_attempts, + uint32_t ack_timeout_ms, uint32_t retry_delay_ms) + : _queue(capacity, max_frame_len), _sender(nullptr), + _max_attempts(max_attempts < 1 ? 1 : max_attempts), _ack_timeout_ms(ack_timeout_ms), + _retry_delay_ms(retry_delay_ms), _pending(nullptr), _pending_len(0), _attempts(0), + _state(STATE_IDLE), _deadline(0), _cb_count(0), _cb_status(0), _cb_consumed(0), _owed(0), + _sent(0), _retries(0), _failed(0), _timeouts(0), _radio_refusals(0) { + _pending = new uint8_t[max_frame_len]; +} + +BridgeTxQueue::~BridgeTxQueue() { + delete[] _pending; +} + +bool BridgeTxQueue::enqueue(const uint8_t *frame, size_t len) { + return _queue.push(frame, len); +} + +void BridgeTxQueue::onSendComplete(bool ok) { + // Runs on the radio driver's task; loop() acts on it from the main task. + _cb_status.store(ok ? 1 : 0, std::memory_order_relaxed); + _cb_count.fetch_add(1, std::memory_order_release); +} + +void BridgeTxQueue::loop(uint32_t now) { + const uint32_t seen = _cb_count.load(std::memory_order_acquire); + uint32_t fresh = seen - _cb_consumed; + _cb_consumed = seen; + + // Completions owed by abandoned attempts say nothing about the frame in flight. + while (fresh > 0 && _owed > 0) { + _owed--; + fresh--; + } + + if (_state == STATE_IN_FLIGHT) { + if (fresh > 0) { + if (_cb_status.load(std::memory_order_relaxed) != 0) { + _sent++; + goIdle(); + } else { + scheduleRetry(now); + } + } else if (hasElapsed(now, _deadline)) { + // Never wait forever, or the bridge goes silent. The radio still owes us + // this attempt's completion, so remember not to believe it later. + _timeouts++; + _owed++; + scheduleRetry(now); + } + return; + } + + if (_state == STATE_IDLE) { + _pending_len = (uint16_t)_queue.pop(_pending, _queue.maxFrameLen()); + if (_pending_len == 0) return; // nothing waiting + + _attempts = 0; + _deadline = now; + _state = STATE_PENDING; + } + + if (_state == STATE_PENDING && hasElapsed(now, _deadline)) { + transmit(now); + } +} + +void BridgeTxQueue::transmit(uint32_t now) { + if (_attempts >= _max_attempts) { + _failed++; // out of attempts; drop it so the frames behind it still move + goIdle(); + return; + } + + _attempts++; + if (_attempts > 1) _retries++; + + if (_sender == nullptr) { + scheduleRetry(now); + return; + } + + _deadline = now + _ack_timeout_ms; + + if (_sender->sendFrame(_pending, _pending_len)) { + _state = STATE_IN_FLIGHT; + } else { + // Refused outright, so no completion is coming. Hold the frame and retry. + _radio_refusals++; + scheduleRetry(now); + } +} + +void BridgeTxQueue::scheduleRetry(uint32_t now) { + _deadline = now + _retry_delay_ms; + _state = STATE_PENDING; +} + +void BridgeTxQueue::goIdle() { + _pending_len = 0; + _attempts = 0; + _state = STATE_IDLE; +} + +void BridgeTxQueue::reset() { + _queue.clear(); + + // An in-flight send still completes after this returns; do not credit it to + // whatever the next session sends first. + if (_state == STATE_IN_FLIGHT) _owed++; + + goIdle(); +} + +void BridgeTxQueue::resetStats() { + _sent = 0; + _retries = 0; + _failed = 0; + _timeouts = 0; + _radio_refusals = 0; + _queue.resetStats(); +} diff --git a/src/helpers/bridges/BridgeTxQueue.h b/src/helpers/bridges/BridgeTxQueue.h new file mode 100644 index 0000000000..3d9b26ad8e --- /dev/null +++ b/src/helpers/bridges/BridgeTxQueue.h @@ -0,0 +1,132 @@ +#pragma once + +#include "BridgeFrameQueue.h" + +#include +#include +#include + +/** + * @brief Sink that hands a framed blob to the radio. + * + * Implemented by the bridge; a fake stands in for it under test. + */ +class BridgeFrameSender { +public: + virtual ~BridgeFrameSender() = default; + + /** + * @return true if the radio accepted the frame and a completion will follow; + * false if it refused it outright (e.g. ESP_ERR_ESPNOW_NO_MEM), in + * which case no completion is coming and the frame must be retried. + */ + virtual bool sendFrame(const uint8_t *frame, size_t len) = 0; +}; + +/** + * @brief Paces bridge transmissions: one frame in flight, retried on failure. + * + * ESP-NOW refuses a send with ESP_ERR_ESPNOW_NO_MEM when its internal queue is + * full, which is what happens when packets arrive in a burst. The frame stays + * queued and is retried, so a burst costs latency instead of messages. + * + * Every path out of the in-flight state is bounded: if the completion callback + * never fires the frame times out, so a lost callback cannot wedge the queue. + */ +class BridgeTxQueue { +public: + /** + * @param capacity frames that can be queued at once + * @param max_frame_len largest frame accepted + * @param max_attempts transmissions tried per frame before giving up + * @param ack_timeout_ms how long to wait for a completion before retrying + * @param retry_delay_ms pause between attempts + */ + BridgeTxQueue(uint8_t capacity, uint16_t max_frame_len, uint8_t max_attempts, + uint32_t ack_timeout_ms, uint32_t retry_delay_ms); + ~BridgeTxQueue(); + + BridgeTxQueue(const BridgeTxQueue &) = delete; + BridgeTxQueue &operator=(const BridgeTxQueue &) = delete; + + void setSender(BridgeFrameSender *sender) { _sender = sender; } + + /** Queue a frame for transmission. False if it was dropped (full/oversize). */ + bool enqueue(const uint8_t *frame, size_t len); + + /** Drive the state machine. Call from the main loop with the current millis(). */ + void loop(uint32_t now); + + /** + * @brief Report the radio's completion. Safe to call from a driver task. + * + * Only stores a flag; the work happens in loop() on the main task. + */ + void onSendComplete(bool ok); + + /** Drop everything queued and in flight, e.g. when the bridge is stopped. */ + void reset(); + + bool isIdle() const { return _state == STATE_IDLE; } + + /** Frames the radio confirmed. */ + uint32_t getSent() const { return _sent; } + /** Transmissions beyond the first attempt for a frame. */ + uint32_t getRetries() const { return _retries; } + /** Frames abandoned after max_attempts. */ + uint32_t getFailed() const { return _failed; } + /** Completions that never arrived within ack_timeout_ms. */ + uint32_t getTimeouts() const { return _timeouts; } + /** Times the radio refused a frame outright. */ + uint32_t getRadioRefusals() const { return _radio_refusals; } + + BridgeFrameQueue &queue() { return _queue; } + const BridgeFrameQueue &queue() const { return _queue; } + + void resetStats(); + +private: + enum State : uint8_t { + STATE_IDLE, ///< nothing loaded + STATE_PENDING, ///< frame loaded, waiting until it is time to (re)transmit + STATE_IN_FLIGHT, ///< handed to the radio, waiting for a completion + }; + + void transmit(uint32_t now); + void scheduleRetry(uint32_t now); + void goIdle(); + + BridgeFrameQueue _queue; + BridgeFrameSender *_sender; + + uint8_t _max_attempts; + uint32_t _ack_timeout_ms; + uint32_t _retry_delay_ms; + + uint8_t *_pending; + uint16_t _pending_len; + uint8_t _attempts; + + State _state; + + // One deadline serves both waiting states, which are mutually exclusive: when + // PENDING it is the time of the next attempt, when IN_FLIGHT the time the + // completion stops being worth waiting for. + uint32_t _deadline; + + // The radio owes exactly one completion per accepted send, but its callback + // carries no token identifying which send it belongs to. Counting completions + // rather than latching a single flag lets loop() tell a completion for the + // current attempt apart from one owed by an attempt already timed out or reset + // away, which would otherwise confirm a frame nothing acknowledged. + std::atomic _cb_count; ///< bumped by the driver task per completion + std::atomic _cb_status; ///< outcome of the most recent completion + uint32_t _cb_consumed; ///< completions loop() has already accounted for + uint32_t _owed; ///< completions still due from abandoned attempts + + uint32_t _sent; + uint32_t _retries; + uint32_t _failed; + uint32_t _timeouts; + uint32_t _radio_refusals; +}; diff --git a/src/helpers/bridges/ESPNowBridge.cpp b/src/helpers/bridges/ESPNowBridge.cpp index 3c094e7c90..a923334214 100644 --- a/src/helpers/bridges/ESPNowBridge.cpp +++ b/src/helpers/bridges/ESPNowBridge.cpp @@ -2,28 +2,42 @@ #include #include +#include #ifdef WITH_ESPNOW_BRIDGE +static const uint8_t BROADCAST_ADDRESS[ESP_NOW_ETH_ALEN] = { 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF }; + // Static member to handle callbacks ESPNowBridge *ESPNowBridge::_instance = nullptr; -// Static callback wrappers -void ESPNowBridge::recv_cb(const uint8_t *mac, const uint8_t *data, int32_t len) { - if (_instance) { - _instance->onDataRecv(mac, data, len); - } +// Static callback wrappers. Both run on the Wi-Fi task, not the Arduino loop +// task, so they must not touch the packet pool, the inbound queue or the +// seen-packet table: none of those are synchronised. +#if defined(ESP_IDF_VERSION_MAJOR) && ESP_IDF_VERSION_MAJOR >= 5 +void ESPNowBridge::recv_cb(const esp_now_recv_info_t *info, const uint8_t *data, int len) { + if (_instance == nullptr || len <= 0) return; + _instance->_rx_frames.push(data, (size_t)len); } +#else +void ESPNowBridge::recv_cb(const uint8_t *mac, const uint8_t *data, int len) { + if (_instance == nullptr || len <= 0) return; + _instance->_rx_frames.push(data, (size_t)len); +} +#endif void ESPNowBridge::send_cb(const uint8_t *mac, esp_now_send_status_t status) { - if (_instance) { - _instance->onDataSent(mac, status); - } + if (_instance == nullptr) return; + _instance->_tx_frames.onSendComplete(status == ESP_NOW_SEND_SUCCESS); } ESPNowBridge::ESPNowBridge(NodePrefs *prefs, mesh::PacketManager *mgr, mesh::RTCClock *rtc) - : BridgeBase(prefs, mgr, rtc), _rx_buffer_pos(0) { + : BridgeBase(prefs, mgr, rtc), _rx_frames(RX_QUEUE_DEPTH, MAX_ESPNOW_PACKET_SIZE), + _tx_frames(TX_QUEUE_DEPTH, MAX_ESPNOW_PACKET_SIZE, TX_MAX_ATTEMPTS, TX_ACK_TIMEOUT_MS, + TX_RETRY_DELAY_MS), + _rx_invalid(0) { _instance = this; + _tx_frames.setSender(this); } void ESPNowBridge::begin() { @@ -31,7 +45,10 @@ void ESPNowBridge::begin() { // Initialize WiFi in station mode WiFi.mode(WIFI_STA); - + + // With power save at its default the radio dozes and misses broadcasts. + esp_wifi_set_ps(WIFI_PS_NONE); + // Set Wi-Fi channel if (esp_wifi_set_channel(_prefs->bridge_channel, WIFI_SECOND_CHAN_NONE) != ESP_OK) { BRIDGE_DEBUG_PRINTLN("Error setting WIFI channel to %d\n", _prefs->bridge_channel); @@ -51,8 +68,10 @@ void ESPNowBridge::begin() { // Add broadcast peer esp_now_peer_info_t peerInfo = {}; memset(&peerInfo, 0, sizeof(peerInfo)); - memset(peerInfo.peer_addr, 0xFF, ESP_NOW_ETH_ALEN); // Broadcast address - peerInfo.channel = _prefs->bridge_channel; + memcpy(peerInfo.peer_addr, BROADCAST_ADDRESS, ESP_NOW_ETH_ALEN); + // 0 = use the interface channel set above. An explicit value that disagreed + // with it would make ESP-NOW hop per send. + peerInfo.channel = 0; peerInfo.encrypt = false; if (esp_now_add_peer(&peerInfo) != ESP_OK) { @@ -67,16 +86,17 @@ void ESPNowBridge::begin() { void ESPNowBridge::end() { BRIDGE_DEBUG_PRINTLN("Stopping...\n"); - // Remove broadcast peer - uint8_t broadcastAddress[] = { 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF }; - if (esp_now_del_peer(broadcastAddress) != ESP_OK) { - BRIDGE_DEBUG_PRINTLN("Error removing broadcast peer\n"); - } + _initialized = false; - // Unregister callbacks + // Unregister before tearing down, so nothing can push into the queues esp_now_register_recv_cb(nullptr); esp_now_register_send_cb(nullptr); + // Remove broadcast peer + if (esp_now_del_peer(BROADCAST_ADDRESS) != ESP_OK) { + BRIDGE_DEBUG_PRINTLN("Error removing broadcast peer\n"); + } + // Deinitialize ESP-NOW if (esp_now_deinit() != ESP_OK) { BRIDGE_DEBUG_PRINTLN("Error deinitializing ESP-NOW\n"); @@ -85,136 +105,110 @@ void ESPNowBridge::end() { // Turn off WiFi WiFi.mode(WIFI_OFF); - // Update bridge state - _initialized = false; + // Drop anything still buffered; it belongs to a radio that no longer exists + _tx_frames.reset(); + _rx_frames.clear(); } void ESPNowBridge::loop() { - // Nothing to do here - ESP-NOW is callback based + if (!_initialized) return; + + uint8_t frame[MAX_ESPNOW_PACKET_SIZE]; + drainRxFrames(_rx_frames, RX_DRAIN_PER_LOOP, frame, sizeof(frame)); + + _tx_frames.loop(millis()); } -void ESPNowBridge::xorCrypt(uint8_t *data, size_t len) { - size_t keyLen = strlen(_prefs->bridge_secret); - for (size_t i = 0; i < len; i++) { - data[i] ^= _prefs->bridge_secret[i % keyLen]; +const uint8_t *ESPNowBridge::unwrapFrame(const uint8_t *frame, size_t frame_len, size_t &blob_len) { + const int payload_len = + BridgeCodec::decode(frame, frame_len, _prefs->bridge_secret, _rx_blob, sizeof(_rx_blob)); + if (payload_len <= 0) { + // Wrong magic, bad checksum, or another network's secret + BRIDGE_DEBUG_PRINTLN("RX invalid frame, len=%d\n", (int)frame_len); + _rx_invalid++; + return nullptr; } + + blob_len = (size_t)payload_len; + return _rx_blob; } -void ESPNowBridge::onDataRecv(const uint8_t *mac, const uint8_t *data, int32_t len) { - // Ignore packets that are too small to contain header + checksum - if (len < (BRIDGE_MAGIC_SIZE + BRIDGE_CHECKSUM_SIZE)) { - BRIDGE_DEBUG_PRINTLN("RX packet too small, len=%d\n", len); +void ESPNowBridge::sendPacket(mesh::Packet *packet) { + // Guard against uninitialized state + if (_initialized == false) { return; } - // Validate total packet size - if (len > MAX_ESPNOW_PACKET_SIZE) { - BRIDGE_DEBUG_PRINTLN("RX packet too large, len=%d\n", len); + // First validate the packet pointer + if (!packet) { + BRIDGE_DEBUG_PRINTLN("TX invalid packet pointer\n"); return; } - // Check packet header magic - uint16_t received_magic = (data[0] << 8) | data[1]; - if (received_magic != BRIDGE_PACKET_MAGIC) { - BRIDGE_DEBUG_PRINTLN("RX invalid magic 0x%04X\n", received_magic); + if (_seen_packets.wasSeen(packet)) { + _tx_duplicates++; return; } - // Make a copy we can decrypt - uint8_t decrypted[MAX_ESPNOW_PACKET_SIZE]; - const size_t encryptedDataLen = len - BRIDGE_MAGIC_SIZE; - memcpy(decrypted, data + BRIDGE_MAGIC_SIZE, encryptedDataLen); - - // Try to decrypt (checksum + payload) - xorCrypt(decrypted, encryptedDataLen); - - // Validate checksum - uint16_t received_checksum = (decrypted[0] << 8) | decrypted[1]; - const size_t payloadLen = encryptedDataLen - BRIDGE_CHECKSUM_SIZE; + // writeTo() can emit up to MAX_TRANS_UNIT bytes, so size the buffer for that + // and check whether it fits a frame afterwards. + uint8_t blob[MAX_TRANS_UNIT + 1]; + const uint16_t blob_len = packet->writeTo(blob); - if (!validateChecksum(decrypted + BRIDGE_CHECKSUM_SIZE, payloadLen, received_checksum)) { - // Failed to decrypt - likely from a different network - BRIDGE_DEBUG_PRINTLN("RX checksum mismatch, rcv=0x%04X\n", received_checksum); + uint8_t frame[MAX_ESPNOW_PACKET_SIZE]; + const int frame_len = + BridgeCodec::encode(blob, blob_len, _prefs->bridge_secret, frame, sizeof(frame)); + if (frame_len < 0) { + BRIDGE_DEBUG_PRINTLN("TX packet too large (payload=%d, max=%d)\n", blob_len, + (int)MAX_PAYLOAD_SIZE); + _tx_oversized++; return; } - BRIDGE_DEBUG_PRINTLN("RX, payload_len=%d\n", payloadLen); - - // Create mesh packet - mesh::Packet *pkt = _instance->_mgr->allocNew(); - if (!pkt) return; - - if (pkt->readFrom(decrypted + BRIDGE_CHECKSUM_SIZE, payloadLen)) { - _instance->onPacketReceived(pkt); + if (_tx_frames.enqueue(frame, (size_t)frame_len)) { + // Mark once queued, not once transmitted: a frame awaiting its turn must + // already suppress the same packet arriving back over the bridge. Marking + // only on success leaves a dropped packet eligible if the mesh resends it. + _seen_packets.markSeen(packet); + BRIDGE_DEBUG_PRINTLN("TX queued, len=%d\n", blob_len); } else { - _instance->_mgr->free(pkt); + BRIDGE_DEBUG_PRINTLN("TX queue full, len=%d\n", blob_len); } } -void ESPNowBridge::onDataSent(const uint8_t *mac_addr, esp_now_send_status_t status) { - // Could add transmission error handling here if needed +bool ESPNowBridge::sendFrame(const uint8_t *frame, size_t len) { + return esp_now_send(BROADCAST_ADDRESS, frame, len) == ESP_OK; } -void ESPNowBridge::sendPacket(mesh::Packet *packet) { - // Guard against uninitialized state - if (_initialized == false) { - return; - } +void ESPNowBridge::onPacketReceived(mesh::Packet *packet) { + handleReceivedPacket(packet); +} - // First validate the packet pointer - if (!packet) { - BRIDGE_DEBUG_PRINTLN("TX invalid packet pointer\n"); - return; - } +void ESPNowBridge::getRxStats(char *dest, size_t dest_size) const { + snprintf(dest, dest_size, + "RX in=%u ok=%u dup=%u bad=%u nopar=%u pool=%u qfull=%u hwm=%u/%u", + (unsigned)_rx_frames.getPushed(), (unsigned)_rx_delivered, (unsigned)_rx_duplicates, + (unsigned)_rx_invalid, (unsigned)_rx_unparsed, (unsigned)_rx_no_packet, + (unsigned)_rx_frames.getDroppedFull(), (unsigned)_rx_frames.getHighWaterMark(), + (unsigned)_rx_frames.capacity()); +} - if (!_seen_packets.wasSeen(packet)) { - _seen_packets.markSeen(packet); - // Create a temporary buffer just for size calculation and reuse for actual writing - uint8_t sizingBuffer[MAX_PAYLOAD_SIZE]; - uint16_t meshPacketLen = packet->writeTo(sizingBuffer); - - // Check if packet fits within our maximum payload size - if (meshPacketLen > MAX_PAYLOAD_SIZE) { - BRIDGE_DEBUG_PRINTLN("TX packet too large (payload=%d, max=%d)\n", meshPacketLen, - MAX_PAYLOAD_SIZE); - return; - } - - uint8_t buffer[MAX_ESPNOW_PACKET_SIZE]; - - // Write magic header (2 bytes) - buffer[0] = (BRIDGE_PACKET_MAGIC >> 8) & 0xFF; - buffer[1] = BRIDGE_PACKET_MAGIC & 0xFF; - - // Write packet payload starting after magic header and checksum - const size_t packetOffset = BRIDGE_MAGIC_SIZE + BRIDGE_CHECKSUM_SIZE; - memcpy(buffer + packetOffset, sizingBuffer, meshPacketLen); - - // Calculate and add checksum (only of the payload) - uint16_t checksum = fletcher16(buffer + packetOffset, meshPacketLen); - buffer[2] = (checksum >> 8) & 0xFF; // High byte - buffer[3] = checksum & 0xFF; // Low byte - - // Encrypt payload and checksum (not including magic header) - xorCrypt(buffer + BRIDGE_MAGIC_SIZE, meshPacketLen + BRIDGE_CHECKSUM_SIZE); - - // Total packet size: magic header + checksum + payload - const size_t totalPacketSize = BRIDGE_MAGIC_SIZE + BRIDGE_CHECKSUM_SIZE + meshPacketLen; - - // Broadcast using ESP-NOW - uint8_t broadcastAddress[] = { 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF }; - esp_err_t result = esp_now_send(broadcastAddress, buffer, totalPacketSize); - - if (result == ESP_OK) { - BRIDGE_DEBUG_PRINTLN("TX, len=%d\n", meshPacketLen); - } else { - BRIDGE_DEBUG_PRINTLN("TX FAILED!\n"); - } - } +void ESPNowBridge::getTxStats(char *dest, size_t dest_size) const { + snprintf(dest, dest_size, + "TX ok=%u dup=%u big=%u rty=%u ref=%u tmo=%u fail=%u qfull=%u hwm=%u/%u", + (unsigned)_tx_frames.getSent(), (unsigned)_tx_duplicates, (unsigned)_tx_oversized, + (unsigned)_tx_frames.getRetries(), (unsigned)_tx_frames.getRadioRefusals(), + (unsigned)_tx_frames.getTimeouts(), (unsigned)_tx_frames.getFailed(), + (unsigned)_tx_frames.queue().getDroppedFull(), + (unsigned)_tx_frames.queue().getHighWaterMark(), + (unsigned)_tx_frames.queue().capacity()); } -void ESPNowBridge::onPacketReceived(mesh::Packet *packet) { - handleReceivedPacket(packet); +void ESPNowBridge::resetStats() { + resetBaseStats(); + _rx_invalid = 0; + _rx_frames.resetStats(); + _tx_frames.resetStats(); } #endif diff --git a/src/helpers/bridges/ESPNowBridge.h b/src/helpers/bridges/ESPNowBridge.h index 431a036b09..2ff9b5d3c9 100644 --- a/src/helpers/bridges/ESPNowBridge.h +++ b/src/helpers/bridges/ESPNowBridge.h @@ -3,6 +3,9 @@ #include "MeshCore.h" #include "esp_now.h" #include "helpers/bridges/BridgeBase.h" +#include "helpers/bridges/BridgeCodec.h" +#include "helpers/bridges/BridgeFrameQueue.h" +#include "helpers/bridges/BridgeTxQueue.h" #ifdef WITH_ESPNOW_BRIDGE @@ -19,16 +22,20 @@ * - Duplicate packet detection using SimpleMeshTables tracking * - Maximum packet size of 250 bytes (ESP-NOW limitation) * - * Packet Structure: + * Packet Structure (see BridgeCodec): * [2 bytes] Magic Header - Used to identify ESPNowBridge packets - * [2 bytes] Fletcher-16 checksum of encrypted payload (calculated over payload only) + * [2 bytes] Fletcher-16 checksum of the payload, encrypted * [246 bytes max] Encrypted payload containing the mesh packet * - * The Fletcher-16 checksum is used to validate packet integrity and detect - * corrupted or tampered packets. It's calculated over the encrypted payload - * and provides a simple but effective way to verify packets are both - * uncorrupted and from the same network (since the checksum is calculated - * after encryption). + * Threading: + * ESP-NOW invokes its callbacks on the Wi-Fi task, not the Arduino loop task. + * The packet pool, inbound queue and seen-packet table have no locking, so the + * callbacks touch none of them: recv_cb() copies the frame into _rx_frames and + * send_cb() bumps a counter. All mesh work happens on the main task, in loop() + * and in sendPacket(), which the mesh calls from logRx()/logTx(). + * + * Every way a frame can be lost has a counter behind `get bridge.rxstats` and + * `get bridge.txstats`, so nothing is dropped silently. * * Configuration: * - Define WITH_ESPNOW_BRIDGE to enable this bridge @@ -39,10 +46,19 @@ * _prefs->bridge_secret values. Packets encrypted with a different key will * fail the checksum validation and be discarded. */ -class ESPNowBridge : public BridgeBase { +class ESPNowBridge : public BridgeBase, public BridgeFrameSender { private: static ESPNowBridge *_instance; - static void recv_cb(const uint8_t *mac, const uint8_t *data, int32_t len); + + // ESP-IDF 5 (Arduino core 3.x, as used by the C6 variants) passes a richer + // info struct to the receive callback. +#if defined(ESP_IDF_VERSION_MAJOR) && ESP_IDF_VERSION_MAJOR >= 5 + static void recv_cb(const esp_now_recv_info_t *info, const uint8_t *data, int len); +#else + // int, not int32_t: esp_now_recv_cb_t spells the length `int`, and on the + // RISC-V targets int32_t is `long int`, so int32_t here fails to convert. + static void recv_cb(const uint8_t *mac, const uint8_t *data, int len); +#endif static void send_cb(const uint8_t *mac, esp_now_send_status_t status); /** @@ -58,48 +74,42 @@ class ESPNowBridge : public BridgeBase { */ static const size_t MAX_ESPNOW_PACKET_SIZE = 250; - /** - * Size constants for packet parsing - */ - static const size_t MAX_PAYLOAD_SIZE = MAX_ESPNOW_PACKET_SIZE - (BRIDGE_MAGIC_SIZE + BRIDGE_CHECKSUM_SIZE); + /** Largest mesh packet blob that still fits an ESP-NOW frame once framed. */ + static const size_t MAX_PAYLOAD_SIZE = MAX_ESPNOW_PACKET_SIZE - BridgeCodec::FRAME_OVERHEAD; - /** Buffer for receiving ESP-NOW packets */ - uint8_t _rx_buffer[MAX_ESPNOW_PACKET_SIZE]; + /** Frames buffered in each direction. */ + static const uint8_t RX_QUEUE_DEPTH = 8; + static const uint8_t TX_QUEUE_DEPTH = 8; - /** Current position in receive buffer */ - size_t _rx_buffer_pos; + /** Received frames turned into mesh packets per loop() call. */ + static const uint8_t RX_DRAIN_PER_LOOP = 4; - /** - * Performs XOR encryption/decryption of data - * Used to isolate different mesh networks - * - * Uses _prefs->bridge_secret as the key in a simple XOR operation. - * The same operation is used for both encryption and decryption. - * While not cryptographically secure, it provides basic network isolation. - * - * @param data Pointer to data to encrypt/decrypt - * @param len Length of data in bytes - */ - void xorCrypt(uint8_t *data, size_t len); + /** Transmissions tried per frame before it is abandoned. */ + static const uint8_t TX_MAX_ATTEMPTS = 3; - /** - * ESP-NOW receive callback - * Called by ESP-NOW when a packet is received - * - * @param mac Source MAC address - * @param data Received data - * @param len Length of received data - */ - void onDataRecv(const uint8_t *mac, const uint8_t *data, int32_t len); + /** How long to wait for the ESP-NOW send callback before assuming it is lost. */ + static const uint32_t TX_ACK_TIMEOUT_MS = 250; + + /** Pause between transmission attempts. */ + static const uint32_t TX_RETRY_DELAY_MS = 20; + + /** Frames handed over by the Wi-Fi task, drained by loop() on the main task. */ + BridgeFrameQueue _rx_frames; + + /** Outbound frames, paced one at a time and retried when the radio refuses. */ + BridgeTxQueue _tx_frames; /** - * ESP-NOW send callback - * Called by ESP-NOW after a transmission attempt - * - * @param mac_addr Destination MAC address - * @param status Transmission status + * Decrypted payload handed to the mesh. A member rather than a local so that + * unwrapFrame() can return it; loop() holds the encrypted frame at the same + * time, so this costs no more than the two stack buffers it replaces. */ - void onDataSent(const uint8_t *mac_addr, esp_now_send_status_t status); + uint8_t _rx_blob[MAX_PAYLOAD_SIZE]; + + uint32_t _rx_invalid; ///< failed magic or checksum: foreign network, or corrupt + + /** Decrypts a received frame; see BridgeBase::unwrapFrame(). */ + const uint8_t *unwrapFrame(const uint8_t *frame, size_t frame_len, size_t &blob_len) override; public: /** @@ -114,7 +124,7 @@ class ESPNowBridge : public BridgeBase { /** * Initializes the ESP-NOW bridge * - * - Configures WiFi in station mode + * - Configures WiFi in station mode with power save disabled * - Initializes ESP-NOW protocol * - Registers callbacks * - Sets up broadcast peer @@ -133,7 +143,9 @@ class ESPNowBridge : public BridgeBase { /** * Main loop handler - * ESP-NOW is callback-based, so this is currently empty + * + * Drains received frames into the mesh and drives the transmit queue. Must be + * called regularly; nothing else moves packets through the bridge. */ void loop() override; @@ -147,11 +159,27 @@ class ESPNowBridge : public BridgeBase { /** * Called when a packet needs to be transmitted via ESP-NOW - * Encrypts and broadcasts the packet if not seen before + * Frames and queues the packet if not seen before * * @param packet The mesh packet to transmit */ void sendPacket(mesh::Packet *packet) override; + + /** Hands one frame to ESP-NOW. Called by the transmit queue. */ + bool sendFrame(const uint8_t *frame, size_t len) override; + + /** + * @brief Writes the receive-side counters into @p dest. + * + * Comparing TX on one node against RX on another shows where frames go missing. + */ + void getRxStats(char *dest, size_t dest_size) const; + + /** @brief Writes the transmit-side counters into @p dest. */ + void getTxStats(char *dest, size_t dest_size) const; + + /** Zeroes every counter, so a measurement can start from a known point. */ + void resetStats(); }; #endif diff --git a/src/helpers/bridges/RS232Bridge.cpp b/src/helpers/bridges/RS232Bridge.cpp index f719d342e4..54fe52a5e1 100644 --- a/src/helpers/bridges/RS232Bridge.cpp +++ b/src/helpers/bridges/RS232Bridge.cpp @@ -1,11 +1,13 @@ #include "RS232Bridge.h" #include +#include #ifdef WITH_RS232_BRIDGE RS232Bridge::RS232Bridge(NodePrefs *prefs, Stream &serial, mesh::PacketManager *mgr, mesh::RTCClock *rtc) - : BridgeBase(prefs, mgr, rtc), _serial(&serial) {} + : BridgeBase(prefs, mgr, rtc), _serial(&serial), _framer(MAX_PAYLOAD_SIZE), + _rx_frames(RX_QUEUE_DEPTH, MAX_PAYLOAD_SIZE) {} void RS232Bridge::begin() { BRIDGE_DEBUG_PRINTLN("Initializing at %d baud...\n", _prefs->bridge_baud); @@ -39,6 +41,10 @@ void RS232Bridge::end() { // Update bridge state _initialized = false; + + // Drop anything half-received; the link is going away + _framer.reset(); + _rx_frames.clear(); } void RS232Bridge::loop() { @@ -47,60 +53,20 @@ void RS232Bridge::loop() { return; } - while (_serial->available()) { - uint8_t b = _serial->read(); - - if (_rx_buffer_pos < 2) { - // Waiting for magic word - if ((_rx_buffer_pos == 0 && b == ((BRIDGE_PACKET_MAGIC >> 8) & 0xFF)) || - (_rx_buffer_pos == 1 && b == (BRIDGE_PACKET_MAGIC & 0xFF))) { - _rx_buffer[_rx_buffer_pos++] = b; - } else { - // Invalid magic byte, reset and start over - _rx_buffer_pos = 0; - // Check if this byte could be the start of a new magic word - if (b == ((BRIDGE_PACKET_MAGIC >> 8) & 0xFF)) { - _rx_buffer[_rx_buffer_pos++] = b; - } - } - } else { - // Reading length, payload, and checksum - _rx_buffer[_rx_buffer_pos++] = b; - - if (_rx_buffer_pos >= 4) { - uint16_t len = (_rx_buffer[2] << 8) | _rx_buffer[3]; - - // Validate length field - if (len > (MAX_TRANS_UNIT + 1)) { - BRIDGE_DEBUG_PRINTLN("RX invalid length %d, resetting\n", len); - _rx_buffer_pos = 0; // Invalid length, reset - continue; - } - - if (_rx_buffer_pos == len + SERIAL_OVERHEAD) { // Full packet received - uint16_t received_checksum = (_rx_buffer[4 + len] << 8) | _rx_buffer[5 + len]; - - if (validateChecksum(_rx_buffer + 4, len, received_checksum)) { - BRIDGE_DEBUG_PRINTLN("RX, len=%d crc=0x%04x\n", len, received_checksum); - mesh::Packet *pkt = _mgr->allocNew(); - if (pkt) { - if (pkt->readFrom(_rx_buffer + 4, len)) { - onPacketReceived(pkt); - } else { - BRIDGE_DEBUG_PRINTLN("RX failed to parse packet\n"); - _mgr->free(pkt); - } - } else { - BRIDGE_DEBUG_PRINTLN("RX failed to allocate packet\n"); - } - } else { - BRIDGE_DEBUG_PRINTLN("RX checksum mismatch, rcv=0x%04x\n", received_checksum); - } - _rx_buffer_pos = 0; // Reset for next packet - } - } + // Keep draining the UART even when the mesh cannot take packets: stalling + // here would let the receive FIFO overrun mid-frame and corrupt it. available() + // takes the UART driver lock, so ask once per loop rather than once per byte. + for (int pending = _serial->available(); pending > 0; pending--) { + const uint16_t payload_len = _framer.offer((uint8_t)_serial->read()); + if (payload_len > 0) { + _rx_frames.push(_framer.payload(), payload_len); } } + + // The framer has already stripped the serial framing, so the queued frames are + // mesh packet blobs and the default unwrapFrame() passes them straight through. + uint8_t payload[MAX_PAYLOAD_SIZE]; + drainRxFrames(_rx_frames, RX_DRAIN_PER_LOOP, payload, sizeof(payload)); } void RS232Bridge::sendPacket(mesh::Packet *packet) { @@ -115,38 +81,53 @@ void RS232Bridge::sendPacket(mesh::Packet *packet) { return; } - if (!_seen_packets.wasSeen(packet)) { - _seen_packets.markSeen(packet); - - uint8_t buffer[MAX_SERIAL_PACKET_SIZE]; - uint16_t len = packet->writeTo(buffer + 4); - - // Check if packet fits within our maximum payload size - if (len > (MAX_TRANS_UNIT + 1)) { - BRIDGE_DEBUG_PRINTLN("TX packet too large (payload=%d, max=%d)\n", len, MAX_TRANS_UNIT + 1); - return; - } + if (_seen_packets.wasSeen(packet)) { + _tx_duplicates++; + return; + } - // Build packet header - buffer[0] = (BRIDGE_PACKET_MAGIC >> 8) & 0xFF; // Magic high byte - buffer[1] = BRIDGE_PACKET_MAGIC & 0xFF; // Magic low byte - buffer[2] = (len >> 8) & 0xFF; // Length high byte - buffer[3] = len & 0xFF; // Length low byte + uint8_t blob[MAX_PAYLOAD_SIZE]; + const uint16_t blob_len = packet->writeTo(blob); - // Calculate checksum over the payload - uint16_t checksum = fletcher16(buffer + 4, len); - buffer[4 + len] = (checksum >> 8) & 0xFF; // Checksum high byte - buffer[5 + len] = checksum & 0xFF; // Checksum low byte + uint8_t frame[MAX_SERIAL_PACKET_SIZE]; + const int frame_len = BridgeSerialFramer::encode(blob, blob_len, frame, sizeof(frame)); + if (frame_len < 0) { + BRIDGE_DEBUG_PRINTLN("TX packet too large (payload=%d, max=%d)\n", blob_len, + (int)MAX_PAYLOAD_SIZE); + _tx_oversized++; + return; + } - // Send complete packet - _serial->write(buffer, len + SERIAL_OVERHEAD); + _seen_packets.markSeen(packet); + _serial->write(frame, (size_t)frame_len); - BRIDGE_DEBUG_PRINTLN("TX, len=%d crc=0x%04x\n", len, checksum); - } + BRIDGE_DEBUG_PRINTLN("TX, len=%d\n", blob_len); } void RS232Bridge::onPacketReceived(mesh::Packet *packet) { handleReceivedPacket(packet); } +void RS232Bridge::getRxStats(char *dest, size_t dest_size) const { + snprintf(dest, dest_size, + "RX in=%u ok=%u dup=%u crc=%u len=%u noise=%u nopar=%u pool=%u qfull=%u hwm=%u/%u", + (unsigned)_framer.getFramesDecoded(), (unsigned)_rx_delivered, + (unsigned)_rx_duplicates, (unsigned)_framer.getChecksumErrors(), + (unsigned)_framer.getLengthErrors(), (unsigned)_framer.getResyncBytes(), + (unsigned)_rx_unparsed, (unsigned)_rx_no_packet, + (unsigned)_rx_frames.getDroppedFull(), (unsigned)_rx_frames.getHighWaterMark(), + (unsigned)_rx_frames.capacity()); +} + +void RS232Bridge::getTxStats(char *dest, size_t dest_size) const { + snprintf(dest, dest_size, "TX dup=%u big=%u", (unsigned)_tx_duplicates, + (unsigned)_tx_oversized); +} + +void RS232Bridge::resetStats() { + resetBaseStats(); + _framer.resetStats(); + _rx_frames.resetStats(); +} + #endif diff --git a/src/helpers/bridges/RS232Bridge.h b/src/helpers/bridges/RS232Bridge.h index 8fc1c22c93..58659cab3d 100644 --- a/src/helpers/bridges/RS232Bridge.h +++ b/src/helpers/bridges/RS232Bridge.h @@ -1,6 +1,8 @@ #pragma once #include "helpers/bridges/BridgeBase.h" +#include "helpers/bridges/BridgeFrameQueue.h" +#include "helpers/bridges/BridgeSerialFramer.h" #include @@ -75,13 +77,8 @@ class RS232Bridge : public BridgeBase { /** * @brief Main loop handler for processing incoming serial data * - * Implements a state machine for packet reception: - * 1. Searches for magic header bytes for packet synchronization - * 2. Reads length field to determine expected packet size - * 3. Validates packet length against maximum allowed size - * 4. Receives complete packet payload and checksum - * 5. Validates Fletcher-16 checksum for data integrity - * 6. Creates mesh packet and forwards if valid + * Drains the UART through BridgeSerialFramer, then turns completed frames into + * mesh packets as pool slots become available. */ void loop() override; @@ -110,39 +107,50 @@ class RS232Bridge : public BridgeBase { */ void onPacketReceived(mesh::Packet *packet) override; -private: /** - * RS232 Protocol Structure: - * - Magic header: 2 bytes (packet identification) - * - Length field: 2 bytes (payload length) - * - Payload: variable bytes (mesh packet data) - * - Checksum: 2 bytes (Fletcher-16 over payload) - * Total overhead: 6 bytes + * @brief Writes the receive-side counters into @p dest. + * + * Line noise and checksum failures look very different from pool exhaustion. */ + void getRxStats(char *dest, size_t dest_size) const; - /** - * @brief The total overhead of the serial protocol in bytes. - * Includes: MAGIC_WORD (2) + LENGTH (2) + CHECKSUM (2) = 6 bytes - */ - static constexpr uint16_t SERIAL_OVERHEAD = BRIDGE_MAGIC_SIZE + BRIDGE_LENGTH_SIZE + BRIDGE_CHECKSUM_SIZE; + /** @brief Writes the transmit-side counters into @p dest. */ + void getTxStats(char *dest, size_t dest_size) const; + + /** Zeroes every counter, so a measurement can start from a known point. */ + void resetStats(); + +private: + /** Largest mesh packet blob this bridge will carry. */ + static constexpr uint16_t MAX_PAYLOAD_SIZE = MAX_TRANS_UNIT + 1; /** * @brief The maximum size of a complete packet on the serial line. * - * This is calculated as the sum of: - * - MAX_TRANS_UNIT + 1 for the maximum mesh packet size - * - SERIAL_OVERHEAD for the framing (magic + length + checksum) + * The framing overhead comes from BridgeSerialFramer, which owns the wire + * layout, so this cannot drift out of step with the encoder. */ - static constexpr uint16_t MAX_SERIAL_PACKET_SIZE = (MAX_TRANS_UNIT + 1) + SERIAL_OVERHEAD; + static constexpr uint16_t MAX_SERIAL_PACKET_SIZE = + MAX_PAYLOAD_SIZE + BridgeSerialFramer::FRAME_OVERHEAD; + + /** Complete frames buffered between the UART reader and the mesh. */ + static constexpr uint8_t RX_QUEUE_DEPTH = 4; + + /** Received frames turned into mesh packets per loop() call. */ + static constexpr uint8_t RX_DRAIN_PER_LOOP = 4; /** Hardware serial port interface */ Stream *_serial; - /** Buffer for building received packets */ - uint8_t _rx_buffer[MAX_SERIAL_PACKET_SIZE]; + /** Decodes the byte stream into complete, checksum-verified frames */ + BridgeSerialFramer _framer; - /** Current position in the receive buffer */ - uint16_t _rx_buffer_pos = 0; + /** + * Frames waiting to become mesh packets. Decoupling the UART reader from packet + * allocation lets an exhausted pool delay a frame instead of destroying it, + * while the reader keeps draining the UART. + */ + BridgeFrameQueue _rx_frames; }; #endif diff --git a/test/test_bridge/main.cpp b/test/test_bridge/main.cpp new file mode 100644 index 0000000000..4d820af774 --- /dev/null +++ b/test/test_bridge/main.cpp @@ -0,0 +1,6 @@ +#include + +int main(int argc, char **argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/test/test_bridge/test_bridge_codec.cpp b/test/test_bridge/test_bridge_codec.cpp new file mode 100644 index 0000000000..d1ffcb8080 --- /dev/null +++ b/test/test_bridge/test_bridge_codec.cpp @@ -0,0 +1,190 @@ +#include + +#include + +#include +#include + +namespace { + +const char *SECRET = "LVSITANOS"; + +std::vector makePayload(size_t len) { + std::vector p(len); + for (size_t i = 0; i < len; i++) { + p[i] = (uint8_t)(i * 7 + 3); + } + return p; +} + +} // namespace + +TEST(BridgeCodec, RoundTripsPayload) { + auto payload = makePayload(64); + uint8_t frame[250]; + + int frame_len = BridgeCodec::encode(payload.data(), payload.size(), SECRET, frame, sizeof(frame)); + ASSERT_EQ(frame_len, (int)(payload.size() + BridgeCodec::FRAME_OVERHEAD)); + + uint8_t out[250]; + int out_len = BridgeCodec::decode(frame, frame_len, SECRET, out, sizeof(out)); + + ASSERT_EQ(out_len, (int)payload.size()); + EXPECT_EQ(memcmp(out, payload.data(), payload.size()), 0); +} + +TEST(BridgeCodec, EncodedFrameStartsWithClearTextMagic) { + auto payload = makePayload(16); + uint8_t frame[250]; + + int frame_len = BridgeCodec::encode(payload.data(), payload.size(), SECRET, frame, sizeof(frame)); + ASSERT_GT(frame_len, 0); + + EXPECT_EQ(frame[0], (BridgeCodec::BRIDGE_PACKET_MAGIC >> 8) & 0xFF); + EXPECT_EQ(frame[1], BridgeCodec::BRIDGE_PACKET_MAGIC & 0xFF); +} + +TEST(BridgeCodec, EncodedPayloadIsNotSentInClear) { + auto payload = makePayload(32); + uint8_t frame[250]; + + int frame_len = BridgeCodec::encode(payload.data(), payload.size(), SECRET, frame, sizeof(frame)); + ASSERT_GT(frame_len, 0); + + EXPECT_NE(memcmp(frame + BridgeCodec::FRAME_OVERHEAD, payload.data(), payload.size()), 0); +} + +// The largest blob Packet::writeTo() can emit is 253 bytes, which does not fit an +// ESP-NOW frame. encode() must refuse it rather than smash the caller's buffer. +TEST(BridgeCodec, EncodeRejectsPayloadTooLargeForCapacityWithoutWritingPastIt) { + auto payload = makePayload(253); + + uint8_t guarded[300]; + memset(guarded, 0xAA, sizeof(guarded)); + const size_t cap = 250; + + int frame_len = BridgeCodec::encode(payload.data(), payload.size(), SECRET, guarded, cap); + + EXPECT_EQ(frame_len, -1); + for (size_t i = 0; i < sizeof(guarded); i++) { + ASSERT_EQ(guarded[i], 0xAA) << "encode() wrote to the buffer at offset " << i; + } +} + +TEST(BridgeCodec, EncodeAcceptsPayloadThatExactlyFillsCapacity) { + auto payload = makePayload(250 - BridgeCodec::FRAME_OVERHEAD); + uint8_t frame[250]; + + int frame_len = BridgeCodec::encode(payload.data(), payload.size(), SECRET, frame, sizeof(frame)); + + ASSERT_EQ(frame_len, 250); + + uint8_t out[250]; + ASSERT_EQ(BridgeCodec::decode(frame, frame_len, SECRET, out, sizeof(out)), (int)payload.size()); + EXPECT_EQ(memcmp(out, payload.data(), payload.size()), 0); +} + +TEST(BridgeCodec, DecodeRejectsWrongMagic) { + auto payload = makePayload(20); + uint8_t frame[250]; + int frame_len = BridgeCodec::encode(payload.data(), payload.size(), SECRET, frame, sizeof(frame)); + ASSERT_GT(frame_len, 0); + + frame[0] ^= 0xFF; + + uint8_t out[250]; + EXPECT_EQ(BridgeCodec::decode(frame, frame_len, SECRET, out, sizeof(out)), -1); +} + +TEST(BridgeCodec, DecodeRejectsCorruptedPayload) { + auto payload = makePayload(20); + uint8_t frame[250]; + int frame_len = BridgeCodec::encode(payload.data(), payload.size(), SECRET, frame, sizeof(frame)); + ASSERT_GT(frame_len, 0); + + frame[frame_len - 1] ^= 0x01; + + uint8_t out[250]; + EXPECT_EQ(BridgeCodec::decode(frame, frame_len, SECRET, out, sizeof(out)), -1); +} + +TEST(BridgeCodec, DecodeRejectsFrameFromANetworkWithADifferentSecret) { + auto payload = makePayload(20); + uint8_t frame[250]; + int frame_len = BridgeCodec::encode(payload.data(), payload.size(), SECRET, frame, sizeof(frame)); + ASSERT_GT(frame_len, 0); + + uint8_t out[250]; + EXPECT_EQ(BridgeCodec::decode(frame, frame_len, "some-other-net", out, sizeof(out)), -1); +} + +TEST(BridgeCodec, DecodeRejectsTruncatedFrame) { + uint8_t frame[3] = { 0xC0, 0x3E, 0x00 }; + uint8_t out[250]; + + EXPECT_EQ(BridgeCodec::decode(frame, sizeof(frame), SECRET, out, sizeof(out)), -1); + EXPECT_EQ(BridgeCodec::decode(frame, 0, SECRET, out, sizeof(out)), -1); +} + +TEST(BridgeCodec, DecodeRejectsFrameCarryingNoPayload) { + uint8_t frame[BridgeCodec::FRAME_OVERHEAD]; + int frame_len = BridgeCodec::encode(nullptr, 0, SECRET, frame, sizeof(frame)); + // An empty payload is never a valid mesh packet, so it must not encode either. + EXPECT_EQ(frame_len, -1); + + uint8_t only_header[4] = { 0xC0, 0x3E, 0x00, 0x00 }; + uint8_t out[250]; + EXPECT_EQ(BridgeCodec::decode(only_header, sizeof(only_header), SECRET, out, sizeof(out)), -1); +} + +TEST(BridgeCodec, DecodeRejectsPayloadLargerThanOutputCapacity) { + auto payload = makePayload(100); + uint8_t frame[250]; + int frame_len = BridgeCodec::encode(payload.data(), payload.size(), SECRET, frame, sizeof(frame)); + ASSERT_GT(frame_len, 0); + + uint8_t small[50]; + EXPECT_EQ(BridgeCodec::decode(frame, frame_len, SECRET, small, sizeof(small)), -1); +} + +// 'set bridge.secret ' with a blank argument leaves an empty string in prefs. +// A modulo by strlen(secret) would divide by zero here. +TEST(BridgeCodec, EmptySecretRoundTripsInsteadOfDividingByZero) { + auto payload = makePayload(20); + uint8_t frame[250]; + + int frame_len = BridgeCodec::encode(payload.data(), payload.size(), "", frame, sizeof(frame)); + ASSERT_EQ(frame_len, (int)(payload.size() + BridgeCodec::FRAME_OVERHEAD)); + + uint8_t out[250]; + ASSERT_EQ(BridgeCodec::decode(frame, frame_len, "", out, sizeof(out)), (int)payload.size()); + EXPECT_EQ(memcmp(out, payload.data(), payload.size()), 0); +} + +TEST(BridgeCodec, NullSecretRoundTripsInsteadOfDereferencingNull) { + auto payload = makePayload(20); + uint8_t frame[250]; + + int frame_len = BridgeCodec::encode(payload.data(), payload.size(), nullptr, frame, sizeof(frame)); + ASSERT_EQ(frame_len, (int)(payload.size() + BridgeCodec::FRAME_OVERHEAD)); + + uint8_t out[250]; + ASSERT_EQ(BridgeCodec::decode(frame, frame_len, nullptr, out, sizeof(out)), (int)payload.size()); + EXPECT_EQ(memcmp(out, payload.data(), payload.size()), 0); +} + +TEST(BridgeCodec, XorCryptIsItsOwnInverse) { + auto payload = makePayload(40); + std::vector buf = payload; + + BridgeCodec::xorCrypt(buf.data(), buf.size(), SECRET); + EXPECT_NE(memcmp(buf.data(), payload.data(), payload.size()), 0); + + BridgeCodec::xorCrypt(buf.data(), buf.size(), SECRET); + EXPECT_EQ(memcmp(buf.data(), payload.data(), payload.size()), 0); +} + +TEST(BridgeCodec, Fletcher16MatchesKnownValue) { + const uint8_t abcde[] = { 'a', 'b', 'c', 'd', 'e' }; + EXPECT_EQ(BridgeCodec::fletcher16(abcde, sizeof(abcde)), 0xC8F0); +} diff --git a/test/test_bridge/test_bridge_frame_queue.cpp b/test/test_bridge/test_bridge_frame_queue.cpp new file mode 100644 index 0000000000..373b45e432 --- /dev/null +++ b/test/test_bridge/test_bridge_frame_queue.cpp @@ -0,0 +1,232 @@ +#include + +#include + +#include +#include + +namespace { + +const uint16_t MAX_FRAME = 250; + +std::vector frameOf(size_t len, uint8_t tag) { + std::vector f(len); + for (size_t i = 0; i < len; i++) { + f[i] = (uint8_t)(tag + i); + } + return f; +} + +} // namespace + +TEST(BridgeFrameQueue, StartsEmpty) { + BridgeFrameQueue q(4, MAX_FRAME); + + EXPECT_TRUE(q.isEmpty()); + EXPECT_EQ(q.count(), 0); + EXPECT_EQ(q.capacity(), 4); +} + +TEST(BridgeFrameQueue, PopReturnsZeroWhenEmpty) { + BridgeFrameQueue q(4, MAX_FRAME); + uint8_t out[MAX_FRAME]; + + EXPECT_EQ(q.pop(out, sizeof(out)), 0u); +} + +TEST(BridgeFrameQueue, RoundTripsASingleFrame) { + BridgeFrameQueue q(4, MAX_FRAME); + auto f = frameOf(37, 0x10); + + ASSERT_TRUE(q.push(f.data(), f.size())); + EXPECT_FALSE(q.isEmpty()); + EXPECT_EQ(q.count(), 1); + + uint8_t out[MAX_FRAME]; + ASSERT_EQ(q.pop(out, sizeof(out)), f.size()); + EXPECT_EQ(memcmp(out, f.data(), f.size()), 0); + EXPECT_TRUE(q.isEmpty()); +} + +TEST(BridgeFrameQueue, PreservesFifoOrder) { + BridgeFrameQueue q(4, MAX_FRAME); + auto a = frameOf(10, 0xA0); + auto b = frameOf(20, 0xB0); + auto c = frameOf(30, 0xC0); + + ASSERT_TRUE(q.push(a.data(), a.size())); + ASSERT_TRUE(q.push(b.data(), b.size())); + ASSERT_TRUE(q.push(c.data(), c.size())); + + uint8_t out[MAX_FRAME]; + ASSERT_EQ(q.pop(out, sizeof(out)), a.size()); + EXPECT_EQ(memcmp(out, a.data(), a.size()), 0); + ASSERT_EQ(q.pop(out, sizeof(out)), b.size()); + EXPECT_EQ(memcmp(out, b.data(), b.size()), 0); + ASSERT_EQ(q.pop(out, sizeof(out)), c.size()); + EXPECT_EQ(memcmp(out, c.data(), c.size()), 0); +} + +TEST(BridgeFrameQueue, AcceptsFramesUpToCapacity) { + BridgeFrameQueue q(3, MAX_FRAME); + auto f = frameOf(8, 0x01); + + EXPECT_TRUE(q.push(f.data(), f.size())); + EXPECT_TRUE(q.push(f.data(), f.size())); + EXPECT_TRUE(q.push(f.data(), f.size())); + EXPECT_EQ(q.count(), 3); +} + +// A full queue means the consumer is not keeping up. Loss is unavoidable there, +// but it must be counted so it is visible instead of silent. +TEST(BridgeFrameQueue, DropsAndCountsFramesPushedWhenFull) { + BridgeFrameQueue q(2, MAX_FRAME); + auto a = frameOf(8, 0xA0); + auto b = frameOf(8, 0xB0); + auto c = frameOf(8, 0xC0); + + ASSERT_TRUE(q.push(a.data(), a.size())); + ASSERT_TRUE(q.push(b.data(), b.size())); + + EXPECT_FALSE(q.push(c.data(), c.size())); + EXPECT_EQ(q.getDroppedFull(), 1u); + EXPECT_EQ(q.count(), 2); + + // the frames already queued survive the overflow + uint8_t out[MAX_FRAME]; + ASSERT_EQ(q.pop(out, sizeof(out)), a.size()); + EXPECT_EQ(memcmp(out, a.data(), a.size()), 0); +} + +TEST(BridgeFrameQueue, RejectsAndCountsOversizeFrames) { + BridgeFrameQueue q(4, 32); + auto big = frameOf(33, 0x55); + + EXPECT_FALSE(q.push(big.data(), big.size())); + EXPECT_EQ(q.getDroppedOversize(), 1u); + EXPECT_TRUE(q.isEmpty()); +} + +TEST(BridgeFrameQueue, AcceptsFrameExactlyAtMaxFrameLen) { + BridgeFrameQueue q(4, 32); + auto exact = frameOf(32, 0x66); + + ASSERT_TRUE(q.push(exact.data(), exact.size())); + + uint8_t out[32]; + EXPECT_EQ(q.pop(out, sizeof(out)), 32u); + EXPECT_EQ(q.getDroppedOversize(), 0u); +} + +TEST(BridgeFrameQueue, RejectsAndCountsEmptyFrames) { + BridgeFrameQueue q(4, MAX_FRAME); + uint8_t dummy = 0; + + EXPECT_FALSE(q.push(&dummy, 0)); + EXPECT_FALSE(q.push(nullptr, 4)); + EXPECT_EQ(q.getDroppedOversize(), 2u); + EXPECT_TRUE(q.isEmpty()); +} + +// If a frame will not fit the consumer's buffer, dropping it keeps the queue +// moving. Leaving it at the head would stall every frame behind it forever. +TEST(BridgeFrameQueue, DiscardsFrameThatDoesNotFitConsumerBufferInsteadOfWedging) { + BridgeFrameQueue q(4, MAX_FRAME); + auto big = frameOf(100, 0x70); + auto small = frameOf(10, 0x80); + + ASSERT_TRUE(q.push(big.data(), big.size())); + ASSERT_TRUE(q.push(small.data(), small.size())); + + uint8_t out[50]; + EXPECT_EQ(q.pop(out, sizeof(out)), 0u); + EXPECT_EQ(q.getDroppedOversize(), 1u); + + // the next frame is still reachable + EXPECT_EQ(q.pop(out, sizeof(out)), small.size()); + EXPECT_EQ(memcmp(out, small.data(), small.size()), 0); +} + +TEST(BridgeFrameQueue, ReusesSlotsAfterDrainingSoItRunsIndefinitely) { + BridgeFrameQueue q(3, MAX_FRAME); + uint8_t out[MAX_FRAME]; + + for (int round = 0; round < 20; round++) { + auto f = frameOf(16, (uint8_t)round); + ASSERT_TRUE(q.push(f.data(), f.size())) << "round " << round; + ASSERT_EQ(q.pop(out, sizeof(out)), f.size()) << "round " << round; + ASSERT_EQ(memcmp(out, f.data(), f.size()), 0) << "round " << round; + } + EXPECT_TRUE(q.isEmpty()); + EXPECT_EQ(q.getDroppedFull(), 0u); +} + +TEST(BridgeFrameQueue, KeepsFramesIntactAcrossRingWrapAround) { + BridgeFrameQueue q(3, MAX_FRAME); + uint8_t out[MAX_FRAME]; + + auto a = frameOf(11, 0xA0); + auto b = frameOf(12, 0xB0); + ASSERT_TRUE(q.push(a.data(), a.size())); + ASSERT_TRUE(q.push(b.data(), b.size())); + ASSERT_EQ(q.pop(out, sizeof(out)), a.size()); + + // head now wraps past the end of the ring while b is still queued + auto c = frameOf(13, 0xC0); + auto d = frameOf(14, 0xD0); + ASSERT_TRUE(q.push(c.data(), c.size())); + ASSERT_TRUE(q.push(d.data(), d.size())); + + ASSERT_EQ(q.pop(out, sizeof(out)), b.size()); + EXPECT_EQ(memcmp(out, b.data(), b.size()), 0); + ASSERT_EQ(q.pop(out, sizeof(out)), c.size()); + EXPECT_EQ(memcmp(out, c.data(), c.size()), 0); + ASSERT_EQ(q.pop(out, sizeof(out)), d.size()); + EXPECT_EQ(memcmp(out, d.data(), d.size()), 0); +} + +TEST(BridgeFrameQueue, TracksHighWaterMark) { + BridgeFrameQueue q(4, MAX_FRAME); + auto f = frameOf(8, 0x01); + uint8_t out[MAX_FRAME]; + + ASSERT_TRUE(q.push(f.data(), f.size())); + ASSERT_TRUE(q.push(f.data(), f.size())); + ASSERT_TRUE(q.push(f.data(), f.size())); + EXPECT_EQ(q.getHighWaterMark(), 3); + + ASSERT_EQ(q.pop(out, sizeof(out)), f.size()); + ASSERT_EQ(q.pop(out, sizeof(out)), f.size()); + EXPECT_EQ(q.getHighWaterMark(), 3) << "high water mark must not fall when the queue drains"; +} + +TEST(BridgeFrameQueue, CountsPushesAndPops) { + BridgeFrameQueue q(4, MAX_FRAME); + auto f = frameOf(8, 0x01); + uint8_t out[MAX_FRAME]; + + ASSERT_TRUE(q.push(f.data(), f.size())); + ASSERT_TRUE(q.push(f.data(), f.size())); + ASSERT_EQ(q.pop(out, sizeof(out)), f.size()); + + EXPECT_EQ(q.getPushed(), 2u); + EXPECT_EQ(q.getPopped(), 1u); +} + +TEST(BridgeFrameQueue, ResetStatsClearsCountersButKeepsQueuedFrames) { + BridgeFrameQueue q(2, MAX_FRAME); + auto f = frameOf(8, 0x01); + + ASSERT_TRUE(q.push(f.data(), f.size())); + ASSERT_TRUE(q.push(f.data(), f.size())); + ASSERT_FALSE(q.push(f.data(), f.size())); + + q.resetStats(); + + EXPECT_EQ(q.getPushed(), 0u); + EXPECT_EQ(q.getPopped(), 0u); + EXPECT_EQ(q.getDroppedFull(), 0u); + EXPECT_EQ(q.getDroppedOversize(), 0u); + EXPECT_EQ(q.getHighWaterMark(), 0); + EXPECT_EQ(q.count(), 2) << "resetting statistics must not discard queued frames"; +} diff --git a/test/test_bridge/test_bridge_serial_framer.cpp b/test/test_bridge/test_bridge_serial_framer.cpp new file mode 100644 index 0000000000..9dfc4b104c --- /dev/null +++ b/test/test_bridge/test_bridge_serial_framer.cpp @@ -0,0 +1,248 @@ +#include + +#include + +#include +#include + +namespace { + +const uint16_t MAX_PAYLOAD = 256; + +const uint8_t MAGIC_HI = (BridgeCodec::BRIDGE_PACKET_MAGIC >> 8) & 0xFF; +const uint8_t MAGIC_LO = BridgeCodec::BRIDGE_PACKET_MAGIC & 0xFF; + +std::vector payloadOf(size_t len, uint8_t tag) { + std::vector p(len); + for (size_t i = 0; i < len; i++) { + p[i] = (uint8_t)(tag + i * 3); + } + return p; +} + +/** Builds a well-formed frame around a payload. */ +std::vector frameOf(const std::vector &payload) { + std::vector f; + f.push_back(MAGIC_HI); + f.push_back(MAGIC_LO); + f.push_back((uint8_t)((payload.size() >> 8) & 0xFF)); + f.push_back((uint8_t)(payload.size() & 0xFF)); + f.insert(f.end(), payload.begin(), payload.end()); + const uint16_t csum = BridgeCodec::fletcher16(payload.data(), payload.size()); + f.push_back((uint8_t)((csum >> 8) & 0xFF)); + f.push_back((uint8_t)(csum & 0xFF)); + return f; +} + +/** Feeds every byte, returning the length reported on the final byte. */ +uint16_t feed(BridgeSerialFramer &fr, const std::vector &bytes) { + uint16_t got = 0; + for (uint8_t b : bytes) { + got = fr.offer(b); + } + return got; +} + +} // namespace + +TEST(BridgeSerialFramer, ReportsNothingUntilAFrameIsComplete) { + BridgeSerialFramer fr(MAX_PAYLOAD); + auto payload = payloadOf(8, 0x10); + auto frame = frameOf(payload); + + for (size_t i = 0; i + 1 < frame.size(); i++) { + ASSERT_EQ(fr.offer(frame[i]), 0u) << "reported a frame early at byte " << i; + } + EXPECT_EQ(fr.offer(frame.back()), payload.size()); +} + +TEST(BridgeSerialFramer, DecodesACompleteFrame) { + BridgeSerialFramer fr(MAX_PAYLOAD); + auto payload = payloadOf(40, 0x20); + + ASSERT_EQ(feed(fr, frameOf(payload)), payload.size()); + EXPECT_EQ(memcmp(fr.payload(), payload.data(), payload.size()), 0); + EXPECT_EQ(fr.getFramesDecoded(), 1u); +} + +TEST(BridgeSerialFramer, DecodesBackToBackFrames) { + BridgeSerialFramer fr(MAX_PAYLOAD); + auto a = payloadOf(10, 0xA0); + auto b = payloadOf(20, 0xB0); + + ASSERT_EQ(feed(fr, frameOf(a)), a.size()); + EXPECT_EQ(memcmp(fr.payload(), a.data(), a.size()), 0); + + ASSERT_EQ(feed(fr, frameOf(b)), b.size()); + EXPECT_EQ(memcmp(fr.payload(), b.data(), b.size()), 0); + EXPECT_EQ(fr.getFramesDecoded(), 2u); +} + +TEST(BridgeSerialFramer, SkipsAndCountsGarbageBeforeTheMagicHeader) { + BridgeSerialFramer fr(MAX_PAYLOAD); + auto payload = payloadOf(12, 0x30); + + std::vector stream = { 0x00, 0x11, 0x22, 0x33 }; + auto frame = frameOf(payload); + stream.insert(stream.end(), frame.begin(), frame.end()); + + ASSERT_EQ(feed(fr, stream), payload.size()); + EXPECT_EQ(memcmp(fr.payload(), payload.data(), payload.size()), 0); + EXPECT_EQ(fr.getResyncBytes(), 4u); +} + +// Noise ending in the first magic byte must not eat the real header that follows. +TEST(BridgeSerialFramer, RecognisesTheHeaderWhenNoiseEndsWithTheFirstMagicByte) { + BridgeSerialFramer fr(MAX_PAYLOAD); + auto payload = payloadOf(6, 0x40); + + std::vector stream = { MAGIC_HI, MAGIC_HI }; + auto frame = frameOf(payload); + // drop the frame's own first magic byte; the second noise byte serves as it + stream.insert(stream.end(), frame.begin() + 1, frame.end()); + + ASSERT_EQ(feed(fr, stream), payload.size()); + EXPECT_EQ(memcmp(fr.payload(), payload.data(), payload.size()), 0); +} + +TEST(BridgeSerialFramer, RejectsAndCountsAZeroLengthFrame) { + BridgeSerialFramer fr(MAX_PAYLOAD); + + std::vector stream = { MAGIC_HI, MAGIC_LO, 0x00, 0x00, 0x00, 0x00 }; + EXPECT_EQ(feed(fr, stream), 0u); + EXPECT_EQ(fr.getLengthErrors(), 1u); + EXPECT_EQ(fr.getFramesDecoded(), 0u); +} + +TEST(BridgeSerialFramer, RejectsAndCountsAnOverlongLengthField) { + BridgeSerialFramer fr(64); + + std::vector stream = { MAGIC_HI, MAGIC_LO, 0x01, 0x00 }; // 256 > 64 + EXPECT_EQ(feed(fr, stream), 0u); + EXPECT_EQ(fr.getLengthErrors(), 1u); +} + +TEST(BridgeSerialFramer, RejectsAndCountsAChecksumMismatch) { + BridgeSerialFramer fr(MAX_PAYLOAD); + auto payload = payloadOf(16, 0x50); + auto frame = frameOf(payload); + frame[6] ^= 0x01; // corrupt a payload byte + + EXPECT_EQ(feed(fr, frame), 0u); + EXPECT_EQ(fr.getChecksumErrors(), 1u); + EXPECT_EQ(fr.getFramesDecoded(), 0u); +} + +// A rejected frame must not wedge the decoder: the next good frame still arrives. +TEST(BridgeSerialFramer, DecodesTheNextFrameAfterARejectedOne) { + BridgeSerialFramer fr(MAX_PAYLOAD); + auto bad = payloadOf(16, 0x50); + auto bad_frame = frameOf(bad); + bad_frame[6] ^= 0x01; + ASSERT_EQ(feed(fr, bad_frame), 0u); + + auto good = payloadOf(24, 0x60); + ASSERT_EQ(feed(fr, frameOf(good)), good.size()); + EXPECT_EQ(memcmp(fr.payload(), good.data(), good.size()), 0); +} + +TEST(BridgeSerialFramer, DecodesTheNextFrameAfterABadLengthField) { + BridgeSerialFramer fr(64); + + std::vector junk = { MAGIC_HI, MAGIC_LO, 0x01, 0x00 }; + ASSERT_EQ(feed(fr, junk), 0u); + + auto good = payloadOf(20, 0x70); + ASSERT_EQ(feed(fr, frameOf(good)), good.size()); + EXPECT_EQ(memcmp(fr.payload(), good.data(), good.size()), 0); +} + +TEST(BridgeSerialFramer, AcceptsAPayloadExactlyAtTheMaximum) { + BridgeSerialFramer fr(64); + auto payload = payloadOf(64, 0x80); + + ASSERT_EQ(feed(fr, frameOf(payload)), payload.size()); + EXPECT_EQ(memcmp(fr.payload(), payload.data(), payload.size()), 0); + EXPECT_EQ(fr.getLengthErrors(), 0u); +} + +TEST(BridgeSerialFramer, ResetDiscardsAPartialFrame) { + BridgeSerialFramer fr(MAX_PAYLOAD); + auto payload = payloadOf(30, 0x90); + auto frame = frameOf(payload); + + for (size_t i = 0; i < 10; i++) { + fr.offer(frame[i]); + } + fr.reset(); + + // the tail of the abandoned frame must not be mistaken for a frame + for (size_t i = 10; i < frame.size(); i++) { + ASSERT_EQ(fr.offer(frame[i]), 0u); + } + + // a fresh frame still decodes + ASSERT_EQ(feed(fr, frameOf(payload)), payload.size()); +} + +TEST(BridgeSerialFramer, ResetStatsClearsCounters) { + BridgeSerialFramer fr(MAX_PAYLOAD); + auto payload = payloadOf(8, 0xA0); + ASSERT_EQ(feed(fr, frameOf(payload)), payload.size()); + ASSERT_EQ(fr.getFramesDecoded(), 1u); + + fr.resetStats(); + + EXPECT_EQ(fr.getFramesDecoded(), 0u); + EXPECT_EQ(fr.getChecksumErrors(), 0u); + EXPECT_EQ(fr.getLengthErrors(), 0u); + EXPECT_EQ(fr.getResyncBytes(), 0u); +} + +TEST(BridgeSerialFramer, EncodeThenDecodeRoundTrips) { + auto payload = payloadOf(50, 0xB0); + uint8_t frame[512]; + + const int frame_len = BridgeSerialFramer::encode(payload.data(), payload.size(), frame, + sizeof(frame)); + ASSERT_EQ(frame_len, (int)(payload.size() + BridgeSerialFramer::FRAME_OVERHEAD)); + + BridgeSerialFramer fr(MAX_PAYLOAD); + uint16_t got = 0; + for (int i = 0; i < frame_len; i++) { + got = fr.offer(frame[i]); + } + ASSERT_EQ(got, payload.size()); + EXPECT_EQ(memcmp(fr.payload(), payload.data(), payload.size()), 0); +} + +TEST(BridgeSerialFramer, EncodeMatchesTheWireLayoutTheDecoderExpects) { + auto payload = payloadOf(4, 0x11); + uint8_t frame[64]; + const int frame_len = BridgeSerialFramer::encode(payload.data(), payload.size(), frame, + sizeof(frame)); + ASSERT_GT(frame_len, 0); + + EXPECT_EQ(frame[0], MAGIC_HI); + EXPECT_EQ(frame[1], MAGIC_LO); + EXPECT_EQ(frame[2], 0x00); + EXPECT_EQ(frame[3], 0x04); + EXPECT_EQ(memcmp(frame + 4, payload.data(), payload.size()), 0); +} + +TEST(BridgeSerialFramer, EncodeRejectsPayloadTooLargeForCapacityWithoutWritingPastIt) { + auto payload = payloadOf(100, 0xC0); + + uint8_t guarded[200]; + memset(guarded, 0xAA, sizeof(guarded)); + + EXPECT_EQ(BridgeSerialFramer::encode(payload.data(), payload.size(), guarded, 100), -1); + for (size_t i = 0; i < sizeof(guarded); i++) { + ASSERT_EQ(guarded[i], 0xAA) << "encode() wrote to the buffer at offset " << i; + } +} + +TEST(BridgeSerialFramer, EncodeRejectsEmptyPayload) { + uint8_t frame[64]; + EXPECT_EQ(BridgeSerialFramer::encode(nullptr, 0, frame, sizeof(frame)), -1); +} diff --git a/test/test_bridge/test_bridge_tx_queue.cpp b/test/test_bridge/test_bridge_tx_queue.cpp new file mode 100644 index 0000000000..2db00db719 --- /dev/null +++ b/test/test_bridge/test_bridge_tx_queue.cpp @@ -0,0 +1,390 @@ +#include + +#include + +#include +#include + +namespace { + +const uint16_t MAX_FRAME = 250; +const uint8_t MAX_ATTEMPTS = 3; +const uint32_t ACK_TIMEOUT_MS = 100; +const uint32_t RETRY_DELAY_MS = 20; + +/** Stands in for the radio. Records every hand-off and can be told to refuse. */ +class FakeSender : public BridgeFrameSender { +public: + std::vector> sent; + bool accept = true; + + bool sendFrame(const uint8_t *frame, size_t len) override { + if (!accept) return false; + sent.push_back(std::vector(frame, frame + len)); + return true; + } +}; + +std::vector frameOf(size_t len, uint8_t tag) { + std::vector f(len); + for (size_t i = 0; i < len; i++) { + f[i] = (uint8_t)(tag + i); + } + return f; +} + +BridgeTxQueue *makeQueue(FakeSender &sender, uint8_t capacity = 4) { + auto *q = new BridgeTxQueue(capacity, MAX_FRAME, MAX_ATTEMPTS, ACK_TIMEOUT_MS, RETRY_DELAY_MS); + q->setSender(&sender); + return q; +} + +} // namespace + +TEST(BridgeTxQueue, SendsNothingWhileTheQueueIsEmpty) { + FakeSender sender; + auto q = makeQueue(sender); + + q->loop(0); + q->loop(1000); + + EXPECT_TRUE(sender.sent.empty()); + EXPECT_TRUE(q->isIdle()); + delete q; +} + +TEST(BridgeTxQueue, SendsAnEnqueuedFrameOnTheNextLoop) { + FakeSender sender; + auto q = makeQueue(sender); + auto f = frameOf(20, 0xA0); + + ASSERT_TRUE(q->enqueue(f.data(), f.size())); + q->loop(0); + + ASSERT_EQ(sender.sent.size(), 1u); + EXPECT_EQ(sender.sent[0], f); + delete q; +} + +TEST(BridgeTxQueue, KeepsOnlyOneFrameInFlightUntilItCompletes) { + FakeSender sender; + auto q = makeQueue(sender); + auto a = frameOf(10, 0xA0); + auto b = frameOf(10, 0xB0); + + ASSERT_TRUE(q->enqueue(a.data(), a.size())); + ASSERT_TRUE(q->enqueue(b.data(), b.size())); + + q->loop(0); + q->loop(1); + q->loop(2); + + EXPECT_EQ(sender.sent.size(), 1u) << "a second frame must not be handed over while one is in flight"; + delete q; +} + +TEST(BridgeTxQueue, SendsTheNextFrameOnceTheRadioConfirms) { + FakeSender sender; + auto q = makeQueue(sender); + auto a = frameOf(10, 0xA0); + auto b = frameOf(10, 0xB0); + + ASSERT_TRUE(q->enqueue(a.data(), a.size())); + ASSERT_TRUE(q->enqueue(b.data(), b.size())); + + q->loop(0); + q->onSendComplete(true); + q->loop(1); + EXPECT_EQ(q->getSent(), 1u); + + q->loop(2); + ASSERT_EQ(sender.sent.size(), 2u); + EXPECT_EQ(sender.sent[0], a); + EXPECT_EQ(sender.sent[1], b); + + q->onSendComplete(true); + q->loop(3); + EXPECT_EQ(q->getSent(), 2u) << "getSent() counts frames the radio confirmed"; + EXPECT_TRUE(q->isIdle()); + delete q; +} + +// esp_now_send() returns ESP_ERR_ESPNOW_NO_MEM under burst load. The old code +// logged that and threw the packet away; it must survive to be retried. +TEST(BridgeTxQueue, RetriesTheSameFrameWhenTheRadioRefusesIt) { + FakeSender sender; + auto q = makeQueue(sender); + auto f = frameOf(20, 0xA0); + + sender.accept = false; + ASSERT_TRUE(q->enqueue(f.data(), f.size())); + q->loop(0); + EXPECT_TRUE(sender.sent.empty()); + EXPECT_EQ(q->getRadioRefusals(), 1u); + + sender.accept = true; + q->loop(RETRY_DELAY_MS); + + ASSERT_EQ(sender.sent.size(), 1u) << "the refused frame must be retried, not dropped"; + EXPECT_EQ(sender.sent[0], f); + EXPECT_EQ(q->getRetries(), 1u); + delete q; +} + +TEST(BridgeTxQueue, WaitsTheRetryDelayBeforeTryingAgain) { + FakeSender sender; + auto q = makeQueue(sender); + auto f = frameOf(20, 0xA0); + + sender.accept = false; + ASSERT_TRUE(q->enqueue(f.data(), f.size())); + q->loop(0); + + sender.accept = true; + q->loop(RETRY_DELAY_MS - 1); + EXPECT_TRUE(sender.sent.empty()) << "must not retry before the backoff has elapsed"; + + q->loop(RETRY_DELAY_MS); + EXPECT_EQ(sender.sent.size(), 1u); + delete q; +} + +TEST(BridgeTxQueue, GivesUpOnAFrameAfterMaxAttemptsAndMovesOn) { + FakeSender sender; + auto q = makeQueue(sender); + auto a = frameOf(10, 0xA0); + auto b = frameOf(10, 0xB0); + + sender.accept = false; + ASSERT_TRUE(q->enqueue(a.data(), a.size())); + ASSERT_TRUE(q->enqueue(b.data(), b.size())); + + uint32_t now = 0; + for (int i = 0; i < MAX_ATTEMPTS; i++) { + q->loop(now); + now += RETRY_DELAY_MS; + } + q->loop(now); + + EXPECT_EQ(q->getFailed(), 1u); + + sender.accept = true; + now += RETRY_DELAY_MS; + q->loop(now); + + ASSERT_EQ(sender.sent.size(), 1u) << "the queue must move on to the next frame"; + EXPECT_EQ(sender.sent[0], b); + delete q; +} + +// If the radio's completion callback is never invoked, nothing else will ever +// clear the in-flight state. Without a timeout the bridge goes permanently silent. +TEST(BridgeTxQueue, RecoversWhenTheCompletionCallbackNeverFires) { + FakeSender sender; + auto q = makeQueue(sender); + auto f = frameOf(20, 0xA0); + + ASSERT_TRUE(q->enqueue(f.data(), f.size())); + q->loop(0); + ASSERT_EQ(sender.sent.size(), 1u); + + // no onSendComplete() ever arrives + q->loop(ACK_TIMEOUT_MS - 1); + EXPECT_EQ(sender.sent.size(), 1u); + EXPECT_EQ(q->getTimeouts(), 0u); + + q->loop(ACK_TIMEOUT_MS); + EXPECT_EQ(q->getTimeouts(), 1u); + + q->loop(ACK_TIMEOUT_MS + RETRY_DELAY_MS); + EXPECT_EQ(sender.sent.size(), 2u) << "the frame must be retried after the completion timed out"; + delete q; +} + +TEST(BridgeTxQueue, DoesNotStayWedgedWhenEveryCompletionIsLost) { + FakeSender sender; + auto q = makeQueue(sender); + auto a = frameOf(10, 0xA0); + auto b = frameOf(10, 0xB0); + + ASSERT_TRUE(q->enqueue(a.data(), a.size())); + ASSERT_TRUE(q->enqueue(b.data(), b.size())); + + for (uint32_t now = 0; now <= 2000; now += 10) { + q->loop(now); + } + + EXPECT_TRUE(q->isIdle()) << "the queue must drain even if no completion ever arrives"; + EXPECT_GT(q->getTimeouts(), 0u); + delete q; +} + +TEST(BridgeTxQueue, RetriesWhenTheRadioReportsTheSendFailed) { + FakeSender sender; + auto q = makeQueue(sender); + auto f = frameOf(20, 0xA0); + + ASSERT_TRUE(q->enqueue(f.data(), f.size())); + q->loop(0); + ASSERT_EQ(sender.sent.size(), 1u); + + q->onSendComplete(false); + q->loop(1); + q->loop(1 + RETRY_DELAY_MS); + + ASSERT_EQ(sender.sent.size(), 2u); + EXPECT_EQ(sender.sent[1], f); + EXPECT_EQ(q->getRetries(), 1u); + delete q; +} + +TEST(BridgeTxQueue, PreservesOrderAcrossRetries) { + FakeSender sender; + auto q = makeQueue(sender); + auto a = frameOf(10, 0xA0); + auto b = frameOf(10, 0xB0); + auto c = frameOf(10, 0xC0); + + ASSERT_TRUE(q->enqueue(a.data(), a.size())); + ASSERT_TRUE(q->enqueue(b.data(), b.size())); + ASSERT_TRUE(q->enqueue(c.data(), c.size())); + + uint32_t now = 0; + // a is refused once, then everything flows + sender.accept = false; + q->loop(now); + sender.accept = true; + + for (int i = 0; i < 10; i++) { + now += RETRY_DELAY_MS; + q->loop(now); + q->onSendComplete(true); + } + + ASSERT_EQ(sender.sent.size(), 3u); + EXPECT_EQ(sender.sent[0], a); + EXPECT_EQ(sender.sent[1], b); + EXPECT_EQ(sender.sent[2], c); + delete q; +} + +TEST(BridgeTxQueue, ReportsWhenAFrameIsDroppedBecauseTheQueueIsFull) { + FakeSender sender; + auto q = makeQueue(sender, 2); + auto f = frameOf(10, 0xA0); + + EXPECT_TRUE(q->enqueue(f.data(), f.size())); + EXPECT_TRUE(q->enqueue(f.data(), f.size())); + EXPECT_FALSE(q->enqueue(f.data(), f.size())); + + EXPECT_EQ(q->queue().getDroppedFull(), 1u); + delete q; +} + +TEST(BridgeTxQueue, ResetDiscardsQueuedAndInFlightFrames) { + FakeSender sender; + auto q = makeQueue(sender); + auto f = frameOf(10, 0xA0); + + ASSERT_TRUE(q->enqueue(f.data(), f.size())); + ASSERT_TRUE(q->enqueue(f.data(), f.size())); + q->loop(0); + ASSERT_EQ(sender.sent.size(), 1u); + + q->reset(); + EXPECT_TRUE(q->isIdle()); + + q->loop(1000); + EXPECT_EQ(sender.sent.size(), 1u) << "nothing queued should survive a reset"; + delete q; +} + +TEST(BridgeTxQueue, SurvivesLoopWithNoSenderConfigured) { + BridgeTxQueue q(4, MAX_FRAME, MAX_ATTEMPTS, ACK_TIMEOUT_MS, RETRY_DELAY_MS); + auto f = frameOf(10, 0xA0); + + ASSERT_TRUE(q.enqueue(f.data(), f.size())); + q.loop(0); + q.loop(1000); + + SUCCEED() << "no crash"; +} + +TEST(BridgeTxQueue, HandlesMillisRollover) { + FakeSender sender; + auto q = makeQueue(sender); + auto f = frameOf(20, 0xA0); + + const uint32_t near_max = 0xFFFFFFFFu - 5; + + ASSERT_TRUE(q->enqueue(f.data(), f.size())); + q->loop(near_max); + ASSERT_EQ(sender.sent.size(), 1u); + + // millis() wraps past zero while the frame is in flight + q->loop(near_max + ACK_TIMEOUT_MS); // wraps + EXPECT_EQ(q->getTimeouts(), 1u) << "the ack timeout must survive a millis() rollover"; + delete q; +} + +// The radio owes exactly one completion per accepted send. If an attempt times +// out, its completion is still coming; crediting it to the retry would confirm a +// frame that was never acknowledged, and would advance the queue past a frame +// whose real completion is still outstanding. +TEST(BridgeTxQueue, IgnoresACompletionThatArrivesAfterItsAttemptTimedOut) { + FakeSender sender; + auto q = makeQueue(sender); + auto a = frameOf(10, 0xA0); + auto b = frameOf(10, 0xB0); + + ASSERT_TRUE(q->enqueue(a.data(), a.size())); + ASSERT_TRUE(q->enqueue(b.data(), b.size())); + + q->loop(0); + ASSERT_EQ(sender.sent.size(), 1u); + + q->loop(ACK_TIMEOUT_MS); + ASSERT_EQ(q->getTimeouts(), 1u); + + q->loop(ACK_TIMEOUT_MS + RETRY_DELAY_MS); + ASSERT_EQ(sender.sent.size(), 2u) << "frame a is retried"; + + // attempt 1's completion finally lands, long after it was given up on + q->onSendComplete(true); + q->loop(ACK_TIMEOUT_MS + RETRY_DELAY_MS + 1); + + EXPECT_EQ(q->getSent(), 0u) << "a stale completion must not confirm the live attempt"; + EXPECT_EQ(sender.sent.size(), 2u) << "and must not advance the queue to the next frame"; + + // attempt 2's real completion still works + q->onSendComplete(true); + q->loop(ACK_TIMEOUT_MS + RETRY_DELAY_MS + 2); + EXPECT_EQ(q->getSent(), 1u); + delete q; +} + +// reset() happens on `set bridge.channel` / `set bridge.secret`, which restart the +// bridge under a send that is still in flight. +TEST(BridgeTxQueue, IgnoresACompletionOwedFromBeforeAReset) { + FakeSender sender; + auto q = makeQueue(sender); + auto a = frameOf(10, 0xA0); + auto b = frameOf(10, 0xB0); + + ASSERT_TRUE(q->enqueue(a.data(), a.size())); + q->loop(0); + ASSERT_EQ(sender.sent.size(), 1u) << "frame a is in flight"; + + q->reset(); + + ASSERT_TRUE(q->enqueue(b.data(), b.size())); + q->loop(1); + ASSERT_EQ(sender.sent.size(), 2u) << "frame b is now in flight"; + + // frame a's completion, dispatched before the reset, arrives now + q->onSendComplete(true); + q->loop(2); + + EXPECT_EQ(q->getSent(), 0u) << "a pre-reset completion must not confirm frame b"; + delete q; +}