diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index 29da2c74..6c7c7060 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -155,6 +155,8 @@ jobs: - name: Verify WiFi, ESP-NOW and combined wireless controls run: | python3 -B test/test_wireless_control.py -v + python3 -B test/test_serial_wifi_sessions.py -v + python3 -B test/test_companion_wifi_session.py -v python3 -B test/test_espnow_radio_lifecycle.py -v - name: Verify message reader buttons, touch targets, and footer layouts diff --git a/docs/companion_radio_full.md b/docs/companion_radio_full.md index a538671b..b1ca8d55 100644 --- a/docs/companion_radio_full.md +++ b/docs/companion_radio_full.md @@ -386,6 +386,23 @@ their views. Companion session state is device-wide, so use one active Companion application at a time. On nRF52, BLE remains available while USB is in terminal or mOTA mode. +On WiFi TCP port 5000, reconnecting from the same IP retains queued outgoing +frames. A different IP takes over immediately, but never receives the previous +IP's replies: up to four queued frames are parked separately for 30 seconds +from takeover. The old IP can reconnect within that window to retrieve them. +Only one displaced IP's backlog is retained; if another client also needs that +slot, the older backlog is discarded so the new connection is not delayed. +This reuses an existing fixed buffer instead of allocating additional frame +storage. Pending commands, signing sessions, and contact streams belonging to +the displaced WiFi session are canceled immediately; already accepted radio +message transmissions are not canceled. Queued frames are retained, not a +resumable command session, and IP matching is not authentication. Turning WiFi +off clears both queues. Stored message history is unaffected. + +Partial TCP writes retain the unsent bytes without interleaving later frames. +After a reconnect, an incomplete frame restarts from its header; the new TCP +connection must not receive only the old connection's trailing bytes. + USB Binary output is queued as complete length-prefixed frames. Temporary CDC or UART backpressure pauses the contact stream; a frame may drain through a smaller hardware FIFO in ordered chunks, but its remainder is retained and no diff --git a/examples/companion_radio/main.cpp b/examples/companion_radio/main.cpp index f5a976cb..d3a9e666 100644 --- a/examples/companion_radio/main.cpp +++ b/examples/companion_radio/main.cpp @@ -1896,6 +1896,14 @@ void halt() { WIFI_DEBUG_PRINTLN("WiFi enabled by BOOT/GPIO 0 control"); } + static void cancelCompanionWiFiSession(void*) { + if (interface_manager.isReplyRouteFor(&wifi_interface)) { + the_mesh.cancelSerialResponseStream(); + } + the_mesh.cancelSerialOperationsForRoute(&wifi_interface); + interface_manager.forgetReplyRouteForDisconnected(&wifi_interface); + } + static void stopCompanionWiFiServices() { if (companion_wifi_services_stopped) return; wifi_interface.end(); @@ -2778,6 +2786,7 @@ void setup() { // add wifi interface #ifdef WIFI_SSID + wifi_interface.setSessionChangedCallback(cancelCompanionWiFiSession, nullptr); #if defined(COMPANION_EXCLUSIVE_WIFI_BLE) if (companionTransportWiFiActiveAtBoot()) { #endif diff --git a/src/helpers/MultiSerialInterface.h b/src/helpers/MultiSerialInterface.h index b2385bb0..e23d3672 100644 --- a/src/helpers/MultiSerialInterface.h +++ b/src/helpers/MultiSerialInterface.h @@ -344,8 +344,12 @@ public: // Other interfaces retain their input until the producer unlocks the route. if (_lockedReplyInterface != nullptr) { if (!isAvailableReplyTarget(_lockedReplyInterface)) return 0; - size_t frameSize = _lockedReplyInterface->checkRecvFrame(dest); - if (frameSize > 0) _lastRxInterface = _lockedReplyInterface; + // A backend may cancel the previous host's stream inside this call + // (e.g. WiFi IP takeover), clearing the lock before returning a new + // host's first command. Preserve the actual source of that command. + BaseSerialInterface* target = _lockedReplyInterface; + size_t frameSize = target->checkRecvFrame(dest); + if (frameSize > 0) _lastRxInterface = target; return frameSize; } diff --git a/src/helpers/esp32/SerialWifiInterface.cpp b/src/helpers/esp32/SerialWifiInterface.cpp index cc432519..6792cc0f 100644 --- a/src/helpers/esp32/SerialWifiInterface.cpp +++ b/src/helpers/esp32/SerialWifiInterface.cpp @@ -2,6 +2,62 @@ #include "../CompanionFrameQueue.h" #include +void SerialWifiInterface::clearBuffers() { + held_queue_len = send_queue_len = 0; + send_offset = 0; + queue_has_ip = false; + held_since = 0; + memset(held_queue, 0, sizeof(held_queue)); + memset(send_queue, 0, sizeof(send_queue)); +} + +void SerialWifiInterface::expireHeldQueue(uint32_t now) { + if (held_queue_len > 0 && (uint32_t)(now - held_since) >= 30000U) { + held_queue_len = 0; + memset(held_queue, 0, sizeof(held_queue)); + } +} + +void SerialWifiInterface::loop() { + expireHeldQueue((uint32_t)millis()); +} + +void SerialWifiInterface::selectClient(const IPAddress& ip, uint32_t now) { + expireHeldQueue(now); + // TCP framing always starts afresh on a new connection, even for the same + // IP. An incomplete outgoing frame is retained whole, not just its suffix. + send_offset = 0; + if (queue_has_ip && queue_ip != ip) { + if (session_changed != nullptr) session_changed(session_context); + + if (held_queue_len > 0 && held_ip == ip) { + // The previous owner returned: restore its queue and park the displaced + // active queue in the same fixed storage. Only one old IP is retained. + for (int i = 0; i < FRAME_QUEUE_SIZE; ++i) { + Frame swap = send_queue[i]; + send_queue[i] = held_queue[i]; + held_queue[i] = swap; + } + const int restored_len = held_queue_len; + held_queue_len = send_queue_len; + send_queue_len = restored_len; + held_ip = queue_ip; + held_since = now; + } else if (send_queue_len > 0) { + // Bounded fallback: a third owner's backlog replaces the older parked + // queue instead of allocating RAM or delaying the new connection. + memcpy(held_queue, send_queue, sizeof(held_queue)); + held_queue_len = send_queue_len; + held_ip = queue_ip; + held_since = now; + send_queue_len = 0; + memset(send_queue, 0, sizeof(send_queue)); + } + } + queue_ip = ip; + queue_has_ip = true; +} + void SerialWifiInterface::begin(int port) { // wifi setup is handled outside of this class, only starts the server server.begin(port); @@ -22,10 +78,16 @@ void SerialWifiInterface::enable() { _isEnabled = true; clearBuffers(); + resetReceivedFrameHeader(); } void SerialWifiInterface::disable() { _isEnabled = false; + deviceConnected = false; + if (client) client.stop(); + if (queue_has_ip && session_changed != nullptr) session_changed(session_context); + resetReceivedFrameHeader(); + clearBuffers(); } size_t SerialWifiInterface::writeFrame(const uint8_t src[], size_t len) { @@ -34,12 +96,17 @@ size_t SerialWifiInterface::writeFrame(const uint8_t src[], size_t len) { return 0; } - if (deviceConnected && len > 0) { - if (!mesh::enqueueCompanionFrame(send_queue, send_queue_len, FRAME_QUEUE_SIZE, - src, len)) { + if (_isEnabled && deviceConnected && src != nullptr && len > 0) { + // Once any bytes of the front frame have reached TCP, priority insertion + // must not move or evict it. New frames may reorder only the unsent tail. + const int pinned = send_offset != 0 ? 1 : 0; + int tail_len = send_queue_len - pinned; + if (!mesh::enqueueCompanionFrame(send_queue + pinned, tail_len, + FRAME_QUEUE_SIZE - pinned, src, len)) { WIFI_DEBUG_PRINTLN("writeFrame(), send_queue is full!"); return 0; } + send_queue_len = tail_len + pinned; return len; } return 0; @@ -50,7 +117,7 @@ bool SerialWifiInterface::isReadBusy() const { } bool SerialWifiInterface::isWriteBusy() const { - return false; + return send_queue_len >= FRAME_QUEUE_SIZE; } bool SerialWifiInterface::hasReceivedFrameHeader() { @@ -63,14 +130,21 @@ void SerialWifiInterface::resetReceivedFrameHeader() { } size_t SerialWifiInterface::checkRecvFrame(uint8_t dest[]) { + expireHeldQueue((uint32_t)millis()); + if (!_isEnabled) return 0; // check if new client connected auto newClient = server.available(); if (newClient) { + const IPAddress new_ip = newClient.remoteIP(); // disconnect existing client deviceConnected = false; client.stop(); + // Partition queues and cancel the displaced session before exposing the + // new connection to response producers or consuming its first command. + selectClient(new_ip, (uint32_t)millis()); + // switch active connection to new client client = newClient; @@ -97,16 +171,22 @@ size_t SerialWifiInterface::checkRecvFrame(uint8_t dest[]) { _last_write = millis(); int len = send_queue[0].len; - uint8_t pkt[3+len]; // use same header as serial interface so client can delimit frames + uint8_t pkt[3 + MAX_FRAME_SIZE]; // serial-compatible framing pkt[0] = '>'; pkt[1] = (len & 0xFF); // LSB pkt[2] = (len >> 8); // MSB memcpy(&pkt[3], send_queue[0].buf, send_queue[0].len); - client.write(pkt, 3 + len); + const size_t total = 3U + len; + const size_t remaining = total - send_offset; + const size_t written = client.write(pkt + send_offset, remaining); + send_offset += written < remaining ? written : remaining; + if (send_offset < total) return 0; + send_offset = 0; send_queue_len--; for (int i = 0; i < send_queue_len; i++) { // delete top item from queue send_queue[i] = send_queue[i + 1]; } + memset(&send_queue[send_queue_len], 0, sizeof(send_queue[send_queue_len])); } else { // check if we are waiting for a frame header diff --git a/src/helpers/esp32/SerialWifiInterface.h b/src/helpers/esp32/SerialWifiInterface.h index b8a84526..bbae9247 100644 --- a/src/helpers/esp32/SerialWifiInterface.h +++ b/src/helpers/esp32/SerialWifiInterface.h @@ -8,7 +8,6 @@ class SerialWifiInterface : public BaseSerialInterface { bool deviceConnected; bool _isEnabled; unsigned long _last_write; - unsigned long adv_restart_time; WiFiServer server; WiFiClient client; @@ -26,12 +25,23 @@ class SerialWifiInterface : public BaseSerialInterface { FrameHeader received_frame_header; #define FRAME_QUEUE_SIZE 4 - int recv_queue_len; - Frame recv_queue[FRAME_QUEUE_SIZE]; + // Reuse the former (unused) receive queue for one displaced client's replies. + // No heap allocation or additional frame buffer is needed on reconnect. + int held_queue_len; + Frame held_queue[FRAME_QUEUE_SIZE]; int send_queue_len; Frame send_queue[FRAME_QUEUE_SIZE]; + IPAddress queue_ip; + IPAddress held_ip; + bool queue_has_ip = false; + uint32_t held_since = 0; + size_t send_offset = 0; + void (*session_changed)(void*) = nullptr; + void* session_context = nullptr; - void clearBuffers() { recv_queue_len = 0; send_queue_len = 0; } + void clearBuffers(); + void expireHeldQueue(uint32_t now); + void selectClient(const IPAddress& ip, uint32_t now); protected: @@ -40,23 +50,31 @@ public: deviceConnected = false; _isEnabled = false; _last_write = 0; - send_queue_len = recv_queue_len = 0; + clearBuffers(); received_frame_header.type = 0; received_frame_header.length = 0; } void begin(int port); void end(); + // Called synchronously while disconnected, before a different IP can send + // commands or receive replies. The owner can cancel route-bound operations. + void setSessionChangedCallback(void (*callback)(void*), void* context) { + session_changed = callback; + session_context = context; + } // BaseSerialInterface methods void enable() override; void disable() override; bool isEnabled() const override { return _isEnabled; } + void loop() override; bool isConnected() const override; bool isReadBusy() const override; bool isWriteBusy() const override; - bool hasPendingIO() const override { return recv_queue_len > 0 || send_queue_len > 0; } + // A parked backlog is not runnable work and must not keep the MCU awake. + bool hasPendingIO() const override { return _isEnabled && deviceConnected && send_queue_len > 0; } size_t writeFrame(const uint8_t src[], size_t len) override; size_t checkRecvFrame(uint8_t dest[]) override; diff --git a/test/fixtures/serial_wifi_sessions/mocks/Arduino.h b/test/fixtures/serial_wifi_sessions/mocks/Arduino.h new file mode 100644 index 00000000..44b34191 --- /dev/null +++ b/test/fixtures/serial_wifi_sessions/mocks/Arduino.h @@ -0,0 +1,8 @@ +#pragma once + +#include +#include +#include + +extern uint32_t mock_millis; +inline unsigned long millis() { return mock_millis; } diff --git a/test/fixtures/serial_wifi_sessions/mocks/WiFi.h b/test/fixtures/serial_wifi_sessions/mocks/WiFi.h new file mode 100644 index 00000000..26b50479 --- /dev/null +++ b/test/fixtures/serial_wifi_sessions/mocks/WiFi.h @@ -0,0 +1,72 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +class IPAddress { + uint32_t _address = 0; +public: + IPAddress() = default; + IPAddress(uint32_t address) : _address(address) {} + IPAddress(uint8_t a, uint8_t b, uint8_t c, uint8_t d) + : _address(uint32_t(a) << 24 | uint32_t(b) << 16 | uint32_t(c) << 8 | d) {} + operator uint32_t() const { return _address; } + bool operator==(const IPAddress& other) const { return _address == other._address; } + bool operator!=(const IPAddress& other) const { return !(*this == other); } +}; + +struct MockSocket { + IPAddress address; + bool connected = true; + size_t write_limit = 0; + size_t read_count = 0; + std::vector sent; + std::deque received; + explicit MockSocket(IPAddress value) : address(value) {} +}; + +class WiFiClient { + std::shared_ptr _socket; +public: + WiFiClient() = default; + explicit WiFiClient(std::shared_ptr socket) : _socket(socket) {} + explicit operator bool() const { return _socket && _socket->connected; } + bool connected() const { return _socket && _socket->connected; } + void stop() { if (_socket) _socket->connected = false; } + IPAddress remoteIP() const { return _socket ? _socket->address : IPAddress(); } + int available() const { return connected() ? int(_socket->received.size()) : 0; } + size_t read(uint8_t* dest, size_t size) { + size_t read = 0; + while (connected() && read < size && !_socket->received.empty()) { + dest[read++] = _socket->received.front(); + _socket->received.pop_front(); + ++_socket->read_count; + } + return read; + } + size_t readBytes(uint8_t* dest, size_t size) { return read(dest, size); } + size_t write(const uint8_t* src, size_t size) { + if (!connected()) return 0; + const size_t written = std::min(size, _socket->write_limit); + _socket->sent.insert(_socket->sent.end(), src, src + written); + return written; + } +}; + +class WiFiServer { +public: + static std::deque incoming; + void begin(int) {} + void end() { incoming.clear(); } + WiFiClient available() { + if (incoming.empty()) return WiFiClient(); + WiFiClient client = incoming.front(); + incoming.pop_front(); + return client; + } +}; diff --git a/test/fixtures/serial_wifi_sessions/test.cpp b/test/fixtures/serial_wifi_sessions/test.cpp new file mode 100644 index 00000000..15c83861 --- /dev/null +++ b/test/fixtures/serial_wifi_sessions/test.cpp @@ -0,0 +1,344 @@ +#include +#include +#include +#include +#include +#include +#include + +uint32_t mock_millis = 0; +std::deque WiFiServer::incoming; + +using Bytes = std::vector; +using Socket = std::shared_ptr; + +static const IPAddress A(192, 168, 1, 20); +static const IPAddress B(192, 168, 1, 30); +static const IPAddress C(192, 168, 1, 40); + +static Bytes framed(std::initializer_list frames) { + Bytes result; + for (const auto& frame : frames) { + result.push_back('>'); + result.push_back(uint8_t(frame.size())); + result.push_back(uint8_t(frame.size() >> 8)); + result.insert(result.end(), frame.begin(), frame.end()); + } + return result; +} + +struct Fixture { + SerialWifiInterface transport; + uint8_t input[MAX_FRAME_SIZE] = {}; + int callbacks = 0; + bool pending_operation = false; + Socket next_client; + + Fixture() { + mock_millis = 100; + WiFiServer::incoming.clear(); + transport.begin(5000); + transport.enable(); + transport.setSessionChangedCallback([](void* context) { + Fixture& fixture = *static_cast(context); + ++fixture.callbacks; + fixture.pending_operation = false; + // Cancellation must happen before a new socket sees an old response or + // has any of its own command bytes consumed. + if (fixture.next_client) { + assert(fixture.next_client->sent.empty()); + assert(fixture.next_client->read_count == 0); + } + }, this); + } + + ~Fixture() { transport.end(); } + + size_t tick() { return transport.checkRecvFrame(input); } + + Socket connect(IPAddress address, const Bytes& command = {}) { + Socket socket = std::make_shared(address); + socket->received.insert(socket->received.end(), command.begin(), command.end()); + next_client = socket; + WiFiServer::incoming.push_back(WiFiClient(socket)); + tick(); + next_client.reset(); + assert(transport.isConnected()); + return socket; + } + + void enqueue(const Bytes& frame) { + assert(transport.writeFrame(frame.data(), frame.size()) == frame.size()); + } + + void drain(const Socket& socket) { + socket->write_limit = std::numeric_limits::max(); + for (int i = 0; i < 12; ++i) tick(); + } +}; + +int main(int argc, char** argv) { + assert(argc == 2); + const int scenario = std::atoi(argv[1]); + Fixture f; + const Bytes reply_a{0x01, 0x21, 0x22, 0x23}; + const Bytes reply_b{0x02, 0x31, 0x32}; + const Bytes reply_c{0x03, 0x41}; + + if (scenario == 0) { + auto old = f.connect(A); + f.enqueue(reply_a); + auto replacement = f.connect(A); + assert(!old->connected && f.callbacks == 0); + f.drain(replacement); + assert(replacement->sent == framed({reply_a})); + } else if (scenario == 1) { + auto old = f.connect(A); + f.enqueue(reply_a); + old->connected = false; + f.tick(); + assert(!f.transport.isConnected()); + mock_millis += 60000; // Same-IP retention is not the different-IP grace timer. + auto replacement = f.connect(A); + f.drain(replacement); + assert(replacement->sent == framed({reply_a}) && f.callbacks == 0); + } else if (scenario == 2) { + f.connect(A); + f.enqueue(reply_a); + auto other = f.connect(B); + f.enqueue(reply_b); + f.drain(other); + assert(other->sent == framed({reply_b}) && f.callbacks == 1); + } else if (scenario == 3) { + f.connect(A); + f.enqueue(reply_a); + auto other = f.connect(B); + f.enqueue(reply_b); + mock_millis += 29999; + auto returned = f.connect(A); + f.drain(returned); + assert(returned->sent == framed({reply_a})); + assert(other->sent.empty()); + auto other_returned = f.connect(B); + f.drain(other_returned); + assert(other_returned->sent == framed({reply_b})); + assert(f.callbacks == 3); + } else if (scenario == 4) { + f.connect(A); + f.enqueue(reply_a); + f.connect(B); + mock_millis += 30000; + auto expired = f.connect(A); + f.drain(expired); + assert(expired->sent.empty()); + } else if (scenario == 5) { + f.connect(A); + f.enqueue(reply_a); + f.connect(B); + mock_millis += 20000; + f.connect(B); // This must not extend A's held deadline. + assert(f.callbacks == 1); + mock_millis += 10000; + auto expired = f.connect(A); + f.drain(expired); + assert(expired->sent.empty()); + } else if (scenario == 6 || scenario == 7) { + mock_millis = UINT32_MAX - 10000; + f.connect(A); + f.enqueue(reply_a); + f.connect(B); + mock_millis += scenario == 6 ? 29999 : 30000; + auto returned = f.connect(A); + f.drain(returned); + assert(returned->sent == (scenario == 6 ? framed({reply_a}) : Bytes{})); + } else if (scenario == 8) { + f.connect(A); + f.enqueue(reply_a); + f.connect(B); + f.enqueue(reply_b); + f.connect(C); // Park B and evict A: only one spare fixed-size queue exists. + f.enqueue(reply_c); + auto evicted = f.connect(A); + f.drain(evicted); + assert(evicted->sent.empty()); + auto newest = f.connect(C); + f.drain(newest); + assert(newest->sent == framed({reply_c})); + } else if (scenario == 9) { + auto socket = f.connect(A); + f.enqueue(reply_a); + socket->write_limit = 2; + for (int i = 0; i < 8; ++i) f.tick(); + assert(socket->sent == framed({reply_a})); + assert(!f.transport.hasPendingIO()); + } else if (scenario == 10) { + auto socket = f.connect(A); + const Bytes push{0x80, 0xA1, 0xA2, 0xA3}; + f.enqueue(push); + socket->write_limit = 2; + f.tick(); + assert(socket->sent.size() == 2); + f.enqueue(reply_a); // Responses have higher queue priority than this push. + f.drain(socket); + assert(socket->sent == framed({push, reply_a})); + } else if (scenario == 11) { + auto old = f.connect(A); + f.enqueue(reply_a); + old->write_limit = 5; // Header and part of payload already sent. + f.tick(); + assert(old->sent.size() == 5); + auto replacement = f.connect(A); + f.drain(replacement); + assert(replacement->sent == framed({reply_a})); + } else if (scenario == 12) { + auto old = f.connect(A); + f.enqueue(reply_a); + old->write_limit = 2; // Partial header must not be a suffix on reconnection. + f.tick(); + auto other = f.connect(B); + f.drain(other); + assert(other->sent.empty()); + auto returned = f.connect(A); + f.drain(returned); + assert(returned->sent == framed({reply_a})); + } else if (scenario == 13 || scenario == 14) { + f.connect(A); + f.enqueue(reply_a); + f.connect(B); + f.enqueue(reply_b); + if (scenario == 13) f.transport.disable(); + else f.transport.end(); + assert(!f.transport.isConnected() && !f.transport.hasPendingIO()); + f.transport.begin(5000); + f.transport.enable(); + auto previous = f.connect(A); + f.drain(previous); + assert(previous->sent.empty()); + auto other = f.connect(B); + f.drain(other); + assert(other->sent.empty()); + } else if (scenario == 15) { + auto old = f.connect(A, Bytes{'<', 4, 0, 0x11}); // Incomplete old command. + assert(old->read_count == 3); + f.pending_operation = true; + f.enqueue(reply_a); + auto other = f.connect(B, Bytes{'<', 2, 0, 0x71, 0x72}); + assert(f.callbacks == 1 && !f.pending_operation); + // The new header is parsed from its start, not attached to the old header. + if (other->read_count == 0) assert(f.tick() == 2); + assert(other->read_count == 5); + assert(f.input[0] == 0x71 && f.input[1] == 0x72); + assert(other->sent.empty()); + } else if (scenario == 16) { + auto old = f.connect(A, Bytes{'<', 4, 0, 0x11}); + assert(old->read_count == 3); + f.pending_operation = true; + auto replacement = f.connect(A, Bytes{'<', 2, 0, 0x61, 0x62}); + assert(f.callbacks == 0 && f.pending_operation); + assert(replacement->read_count == 5); + assert(f.input[0] == 0x61 && f.input[1] == 0x62); + } else if (scenario == 17) { + auto socket = f.connect(A); + f.enqueue(reply_a); + f.tick(); // A zero-byte socket write is backpressure, not frame completion. + assert(socket->sent.empty() && f.transport.hasPendingIO()); + f.drain(socket); + assert(socket->sent == framed({reply_a})); + } else if (scenario == 18) { + f.connect(A); + for (int i = 0; i < 4; ++i) f.enqueue(Bytes{0x01, uint8_t(i)}); + assert(f.transport.writeFrame(reply_a.data(), reply_a.size()) == 0); + f.connect(B); + for (int i = 0; i < 4; ++i) f.enqueue(Bytes{0x02, uint8_t(i)}); + auto returned = f.connect(A); + f.drain(returned); + assert(returned->sent == framed({{0x01, 0}, {0x01, 1}, {0x01, 2}, {0x01, 3}})); + auto other = f.connect(B); + f.drain(other); + assert(other->sent == framed({{0x02, 0}, {0x02, 1}, {0x02, 2}, {0x02, 3}})); + } else if (scenario == 19) { + f.connect(A); + f.enqueue(reply_a); + f.connect(B); + mock_millis += 20000; + f.connect(C); // B has no backlog: preserve A without extending its deadline. + mock_millis += 9999; + auto returned = f.connect(A); + f.drain(returned); + assert(returned->sent == framed({reply_a})); + } else if (scenario == 20) { + f.connect(A); + f.enqueue(reply_a); + f.connect(B); + mock_millis += 20000; + f.connect(C); + assert(!f.transport.hasPendingIO()); // Parked-only data must not busy-loop. + mock_millis += 10000; + f.transport.loop(); + auto expired = f.connect(A); + f.drain(expired); + assert(expired->sent.empty()); + } else if (scenario == 21) { + auto socket = f.connect(A); + const Bytes push{0x80, 0xA1, 0xA2, 0xA3}; + const Bytes keep{0x81, 0xB1}; + const Bytes evict{0x88, 0xC1}; + f.enqueue(push); + socket->write_limit = 2; + f.tick(); + f.enqueue(keep); + f.enqueue(evict); + f.enqueue(reply_a); + f.enqueue(reply_b); // Full queue can evict a waiting push, never its sent head. + f.drain(socket); + assert(socket->sent == framed({push, reply_a, reply_b, keep})); + } else if (scenario == 22) { + f.connect(A, Bytes{'<', 4, 0, 0x11}); + f.pending_operation = true; + f.transport.disable(); + assert(f.callbacks == 1 && !f.pending_operation); + f.transport.end(); // Repeated teardown must not cancel the same owner twice. + assert(f.callbacks == 1); + f.transport.begin(5000); + f.transport.enable(); + auto replacement = f.connect(A, Bytes{'<', 2, 0, 0x61, 0x62}); + assert(replacement->read_count == 5); + assert(f.input[0] == 0x61 && f.input[1] == 0x62); + } else if (scenario == 23) { + auto socket = f.connect(A); + Bytes maximum(MAX_FRAME_SIZE); + for (size_t i = 0; i < maximum.size(); ++i) maximum[i] = uint8_t(i); + maximum[0] = 0x80; // Low-priority push, pinned once any bytes are sent. + f.enqueue(maximum); + socket->write_limit = MAX_FRAME_SIZE + 2; // All but the final payload byte. + f.tick(); + assert(socket->sent.size() == MAX_FRAME_SIZE + 2); + assert(f.transport.hasPendingIO()); + f.enqueue(reply_a); + f.drain(socket); + assert(socket->sent == framed({maximum, reply_a})); + assert(!f.transport.hasPendingIO()); + } else if (scenario == 24) { + auto old = f.connect(A); + Bytes maximum(MAX_FRAME_SIZE); + for (size_t i = 0; i < maximum.size(); ++i) maximum[i] = uint8_t(i ^ 0x55); + maximum[0] = 0x01; + f.enqueue(maximum); + old->write_limit = MAX_FRAME_SIZE + 2; + f.tick(); + assert(old->sent.size() == MAX_FRAME_SIZE + 2); + f.connect(B); + Bytes other_maximum(MAX_FRAME_SIZE, 0xFE); + other_maximum[0] = 0x02; + f.enqueue(other_maximum); + auto returned = f.connect(A); + f.drain(returned); + assert(returned->sent == framed({maximum})); + auto other_returned = f.connect(B); + f.drain(other_returned); + assert(other_returned->sent == framed({other_maximum})); + } else { + assert(false && "unknown scenario"); + } + std::printf("PASS: Wi-Fi session scenario %d\n", scenario); +} diff --git a/test/test_companion_wifi_session.py b/test/test_companion_wifi_session.py new file mode 100644 index 00000000..2cccd713 --- /dev/null +++ b/test/test_companion_wifi_session.py @@ -0,0 +1,154 @@ +"""Exercise WiFi session cancellation against production route ownership code.""" +from pathlib import Path +import os +import subprocess +import tempfile +import unittest + +from test_replay_reset_integration import extract_braced + + +ROOT = Path(__file__).resolve().parents[1] +MAIN = ROOT / "examples/companion_radio/main.cpp" +MESH = ROOT / "examples/companion_radio/MyMesh.cpp" + + +HARNESS = r''' +#include +#include +#include +#include +#include +#include + +struct FakeInterface : BaseSerialInterface { + bool enabled = false; + bool connected = true; + std::deque> input; + void enable() override { enabled = true; } + void disable() override { enabled = false; } + bool isEnabled() const override { return enabled; } + bool isConnected() const override { return connected; } + bool isReadBusy() const override { return false; } + bool isWriteBusy() const override { return false; } + size_t writeFrame(const uint8_t[], size_t len) override { return len; } + size_t checkRecvFrame(uint8_t dest[]) override { + if (input.empty()) return 0; + auto frame = input.front(); input.pop_front(); + memcpy(dest, frame.data(), frame.size()); + return frame.size(); + } +} wifi_interface, usb_interface, bluetooth_interface; +MultiSerialInterface interface_manager; + +constexpr int EXPECTED_ACK_TABLE_SIZE = 6; +struct MyMesh { + bool streaming = false; + int stream_cancels = 0, pending_cancels = 0, radio_cancels = 0; + int trace_cancels = 0, signing_cancels = 0, expirations = 0; + BaseSerialInterface* pending_serial_reply_route = nullptr; + BaseSerialInterface* command_radio_reply_route = nullptr; + BaseSerialInterface* binary_trace_reply_route = nullptr; + BaseSerialInterface* sign_data_reply_route = nullptr; + struct Ack { BaseSerialInterface* reply_route; bool radio_retry; }; + Ack expected_ack_table[EXPECTED_ACK_TABLE_SIZE] = {}; + void cancelSerialResponseStream() { + ++stream_cancels; + if (streaming) { streaming = false; interface_manager.unlockReplyRoute(); } + } + void clearPendingReqs() { ++pending_cancels; pending_serial_reply_route = nullptr; } + void cancelPendingRadioParamApply() { ++radio_cancels; command_radio_reply_route = nullptr; } + void clearBinaryTraceReply() { ++trace_cancels; binary_trace_reply_route = nullptr; } + void cancelSigningSession() { ++signing_cancels; sign_data_reply_route = nullptr; } + void expireExpectedAcks() { ++expirations; } + void cancelSerialOperationsForRoute(BaseSerialInterface*); +} the_mesh; + +@CANCEL_OPERATIONS@ +@CANCEL_SESSION@ + +int main() { + assert(interface_manager.addInterface(InterfaceType::WiFi, &wifi_interface)); + assert(interface_manager.addInterface(InterfaceType::USB, &usb_interface)); + assert(interface_manager.addInterface(InterfaceType::Bluetooth, &bluetooth_interface)); + FakeInterface* routes[] = {&wifi_interface, &usb_interface, &bluetooth_interface}; + for (auto owner : routes) { + interface_manager.enable(); + the_mesh = MyMesh(); + owner->input.push_back({0x04}); + uint8_t command[MAX_FRAME_SIZE] = {}; + assert(interface_manager.checkRecvFrame(command) == 1); + interface_manager.lockReplyRoute(); + the_mesh.streaming = true; + the_mesh.pending_serial_reply_route = owner; + the_mesh.command_radio_reply_route = owner; + the_mesh.binary_trace_reply_route = owner; + the_mesh.sign_data_reply_route = owner; + for (unsigned i = 0; i < EXPECTED_ACK_TABLE_SIZE; ++i) { + the_mesh.expected_ack_table[i] = {routes[i % 3], true}; + } + + // The transport marks itself disconnected during the synchronous callback. + wifi_interface.connected = false; + cancelCompanionWiFiSession(nullptr); + wifi_interface.connected = true; + const bool cancelled = owner == &wifi_interface; + assert(the_mesh.stream_cancels == int(cancelled)); + assert(the_mesh.streaming == !cancelled); + assert(the_mesh.pending_cancels == int(cancelled)); + assert(the_mesh.radio_cancels == int(cancelled)); + assert(the_mesh.trace_cancels == int(cancelled)); + assert(the_mesh.signing_cancels == int(cancelled)); + assert(the_mesh.pending_serial_reply_route == (cancelled ? nullptr : owner)); + assert(the_mesh.command_radio_reply_route == (cancelled ? nullptr : owner)); + assert(the_mesh.binary_trace_reply_route == (cancelled ? nullptr : owner)); + assert(the_mesh.sign_data_reply_route == (cancelled ? nullptr : owner)); + assert(!interface_manager.isReplyRouteFor(&wifi_interface)); + if (!cancelled) assert(interface_manager.isReplyRouteFor(owner)); + for (unsigned i = 0; i < EXPECTED_ACK_TABLE_SIZE; ++i) { + // Detach WiFi's notification, not the radio's accepted message/retry. + assert(the_mesh.expected_ack_table[i].reply_route == + (i % 3 == 0 ? nullptr : routes[i % 3])); + assert(the_mesh.expected_ack_table[i].radio_retry); + } + assert(the_mesh.expirations == 1); + } +} +''' + + +class CompanionWiFiSessionTest(unittest.TestCase): + def test_callback_is_registered_before_wifi_interface_can_start(self): + source = MAIN.read_text(encoding="utf-8") + registration = source.index( + "wifi_interface.setSessionChangedCallback(cancelCompanionWiFiSession, nullptr)" + ) + setup = source.index("void setup()") + add = source.index("interface_manager.addInterface(InterfaceType::WiFi", setup) + start = source.index("startCompanionWiFi();", add) + self.assertLess(registration, add) + self.assertLess(add, start) + + def test_session_reset_cancels_only_wifi_owned_work(self): + main = MAIN.read_text(encoding="utf-8") + mesh = MESH.read_text(encoding="utf-8") + callback = extract_braced(main, "static void cancelCompanionWiFiSession(") + operations = extract_braced(mesh, "void MyMesh::cancelSerialOperationsForRoute(") + source = HARNESS.replace("@CANCEL_OPERATIONS@", operations).replace( + "@CANCEL_SESSION@", callback + ) + with tempfile.TemporaryDirectory() as directory: + executable = Path(directory) / ("wifi_session.exe" if os.name == "nt" else "wifi_session") + result = subprocess.run( + [os.environ.get("CXX", "g++"), "-std=c++17", "-Werror", + f"-I{ROOT / 'test/mocks'}", f"-I{ROOT / 'src'}", + "-x", "c++", "-", "-o", str(executable)], + input=source, text=True, capture_output=True, + ) + self.assertEqual(result.returncode, 0, result.stderr) + result = subprocess.run([str(executable)], text=True, capture_output=True) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_serial_mode_switch/test_serial_mode_switch.cpp b/test/test_serial_mode_switch/test_serial_mode_switch.cpp index f4f8a58e..2ef0beb3 100644 --- a/test/test_serial_mode_switch/test_serial_mode_switch.cpp +++ b/test/test_serial_mode_switch/test_serial_mode_switch.cpp @@ -448,6 +448,54 @@ TEST(MultiSerialInterface, ForgetsOnlyTheDisconnectedHostReplyRoute) { EXPECT_TRUE(manager.isReplyRouteFor(&bluetooth)); } +TEST(MultiSerialInterface, SessionChangeDuringLockedReadKeepsNewFrameOnItsTransport) { + MultiSerialInterface manager; + class ReconnectingInterface : public FakeSerialInterface { + MultiSerialInterface& manager; + public: + bool change_session = false; + explicit ReconnectingInterface(MultiSerialInterface& owner) : manager(owner) {} + size_t checkRecvFrame(uint8_t dest[]) override { + if (change_session) { + change_session = false; + // WiFi's new-IP callback cancels the old contact iterator and then + // forgets its route while the manager is still inside this read. + manager.unlockReplyRoute(); + manager.forgetReplyRouteForDisconnected(this); + } + return FakeSerialInterface::checkRecvFrame(dest); + } + } wifi(manager); + FakeSerialInterface usb, bluetooth; + wifi.connected = usb.connected = bluetooth.connected = true; + ASSERT_TRUE(manager.addInterface(InterfaceType::WiFi, &wifi)); + ASSERT_TRUE(manager.addInterface(InterfaceType::USB, &usb)); + ASSERT_TRUE(manager.addInterface(InterfaceType::Bluetooth, &bluetooth)); + manager.enable(); + + wifi.received_frames.push_back({0x04}); + uint8_t command[MAX_FRAME_SIZE] = {}; + ASSERT_EQ(manager.checkRecvFrame(command), 1u); + manager.lockReplyRoute(); + + wifi.change_session = true; + wifi.received_frames.push_back({0x16}); + ASSERT_EQ(manager.checkRecvFrame(command), 1u); + ASSERT_EQ(command[0], 0x16); + ASSERT_TRUE(manager.isReplyRouteFor(&wifi)); + const uint8_t device_info[] = {0x0D, 0x43}; + ASSERT_EQ(manager.writeFrame(device_info, sizeof(device_info)), sizeof(device_info)); + ASSERT_EQ(wifi.sent_frames.size(), 1u); + EXPECT_TRUE(usb.sent_frames.empty()); + EXPECT_TRUE(bluetooth.sent_frames.empty()); + + // Cancelling the prior WiFi iterator also releases its lock, so another + // transport may dispatch its own command on the next pass. + usb.received_frames.push_back({0x16}); + ASSERT_EQ(manager.checkRecvFrame(command), 1u); + EXPECT_TRUE(manager.isReplyRouteFor(&usb)); +} + TEST(MultiSerialInterface, CapturedAsyncReplySurvivesLaterRouteChanges) { MultiSerialInterface manager; FakeSerialInterface usb; diff --git a/test/test_serial_wifi_sessions.py b/test/test_serial_wifi_sessions.py new file mode 100644 index 00000000..9740ce85 --- /dev/null +++ b/test/test_serial_wifi_sessions.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +"""Exercise production Wi-Fi transport reconnect ownership and partial writes.""" + +from pathlib import Path +import os +import shutil +import subprocess +import sys +import tempfile +import unittest + + +ROOT = Path(__file__).resolve().parents[1] +FIXTURE = ROOT / "test/fixtures/serial_wifi_sessions" +# MinGW does not ship these runtimes. Non-PIE avoids ASan address-space +# collisions in Linux hosts, matching the other native regression harnesses. +SANITIZER_FLAGS = ( + ["-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie"] + if sys.platform.startswith("linux") else [] +) + +SCENARIOS = ( + "same IP live replacement keeps replies", + "same IP reconnect after disconnect keeps replies", + "different IP receives only its own replies", + "old IP returns before grace deadline and swaps queues", + "old IP expires at exactly 30 seconds", + "same IP reconnect does not extend another IP grace deadline", + "old IP returns before deadline across millis rollover", + "old IP expires across millis rollover", + "third IP evicts older parked backlog", + "partial writes retain every byte", + "higher priority reply cannot interrupt partial frame", + "same IP reconnect replays whole partial frame", + "different IP return replays whole partial frame", + "disable clears both queues and active connection", + "end clears both queues and active connection", + "different IP callback precedes new input and resets partial RX", + "same IP preserves pending operations but resets partial RX", + "zero-byte write retains pending frame", + "both bounded queues retain four complete responses", + "empty different IP takeover preserves existing held backlog", + "empty takeover does not extend expiry and parked data does not busy-loop", + "full queue admission cannot evict a partly sent frame", + "disable cancels owner once and resets partial RX state", + "maximum-sized partly sent frame stays intact before higher priority response", + "maximum-sized partial frame and parked queue survive owner swap", +) + + +class SerialWifiSessionTests(unittest.TestCase): + def test_production_transport_sessions(self): + compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++") + if compiler is None: + self.skipTest("a host C++17 compiler is required") + with tempfile.TemporaryDirectory(prefix="meshcore-wifi-sessions-") as directory: + binary = Path(directory) / "wifi-sessions.exe" + compiled = subprocess.run( + [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", + *SANITIZER_FLAGS, + f"-I{FIXTURE / 'mocks'}", f"-I{ROOT / 'src'}", + str(FIXTURE / "test.cpp"), + str(ROOT / "src/helpers/esp32/SerialWifiInterface.cpp"), + "-o", str(binary)], + capture_output=True, text=True, timeout=60, + ) + self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr) + for scenario, description in enumerate(SCENARIOS): + with self.subTest(scenario=description): + checked = subprocess.run( + [str(binary), str(scenario)], capture_output=True, + text=True, timeout=10, + ) + self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr) + self.assertIn(f"PASS: Wi-Fi session scenario {scenario}", checked.stdout) + + +if __name__ == "__main__": + unittest.main()