diff --git a/docs/cli_commands.md b/docs/cli_commands.md index 6e96fcd7..cba6a900 100644 --- a/docs/cli_commands.md +++ b/docs/cli_commands.md @@ -432,7 +432,7 @@ request. `reg=1` confirms the interface is registered for Companion commands. `client=1` means a TCP client is connected. `acc`, `rx`, and `tx` count accepted clients, complete received frames, and complete transmitted frames since boot; the value after `/` is the last frame's hexadecimal opcode. `fail`, `err`, and -`retry` show listener failures, the last socket error, and seconds until retry. +`retry` show listener failures, the last listener-start socket error, and seconds until retry. `drop` gives the last client-disconnect reason. These counters contain no message contents or credentials. An open listener does not prove a router or guest-network firewall permits access from the phone. @@ -442,6 +442,15 @@ reconnect WiFi, interrupt BLE/WebConfig/OTA, replace a healthy app connection, or re-enable explicitly disabled WiFi. `get display.wifi` includes `tcp=1` only when the listener is actually open; `services=1` alone is not proof. +The ESP32 TCP transport services input even while radio logs or other outgoing +frames are pending, so unsolicited traffic cannot starve the app handshake. +Socket sends are non-blocking. A client that makes no outbound progress for +10 seconds while data is pending is disconnected (`drop=tx-timeout`), without +blocking the other transports. `drop=tx-error` identifies a fatal send error. +Commands wait when all reply slots hold delivery-required data, rather than +executing with nowhere to return their reply. A partly sent frame stays pinned +until it finishes; input and output remain separate TCP streams. + See [local maintenance commands](terminal_chat_cli.md#local-maintenance-commands) for the complete access rules. WiFi-only Companions accept all three modes. A Full Companion that runs BLE and infrastructure WiFi diff --git a/docs/companion_radio_full.md b/docs/companion_radio_full.md index cc260aba..18555b29 100644 --- a/docs/companion_radio_full.md +++ b/docs/companion_radio_full.md @@ -536,6 +536,15 @@ frames, and errors without a PC port probe. `get display.wifi` also shows the actual `tcp` listener flag independently of the network-services flag. A listening socket alone cannot detect LAN firewall or guest-network isolation. +ESP32 TCP sends are non-blocking, and inbound commands are serviced alongside +outgoing data rather than waiting for an empty transmit queue. Unsolicited +radio logs therefore cannot keep the app handshake unread. A live client with +pending output is disconnected after 10 seconds without write progress; +other transports continue running. Required replies retain their queue slots, +so a new command waits if no reply can be admitted. Partially sent frames +remain pinned to preserve framing, while already queued replies keep the same +reconnection rules below. + 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 diff --git a/src/helpers/wifi/SerialWifiInterface.cpp b/src/helpers/wifi/SerialWifiInterface.cpp index f827f78f..85f8418b 100644 --- a/src/helpers/wifi/SerialWifiInterface.cpp +++ b/src/helpers/wifi/SerialWifiInterface.cpp @@ -2,12 +2,18 @@ #include "../CompanionFrameQueue.h" #include #include +#if defined(ESP32) +#include +#endif static constexpr uint32_t WIFI_FRAME_TIMEOUT_MS = 5000; +static constexpr uint32_t WIFI_TX_STALL_TIMEOUT_MS = 10000; void SerialWifiInterface::clearBuffers() { held_queue_len = send_queue_len = 0; send_offset = 0; + tx_waiting = false; + tx_last_progress = 0; queue_has_ip = false; held_since = 0; memset(held_queue, 0, sizeof(held_queue)); @@ -27,9 +33,10 @@ void SerialWifiInterface::loop() { // Use enable intent, not isEnabled(): an unavailable listener must recover // even though MultiSerialInterface correctly declines to dispatch into it. if (_isEnabled && listener_requested) serviceListener(now); + updateReceiveCapacityPause(now); if (deviceConnected && !client.connected()) { disconnectClient(true, DisconnectReason::PeerClosed); - } else if (deviceConnected && receive_pending + } else if (deviceConnected && receive_pending && !receive_capacity_paused && (uint32_t)(now - receive_started) >= WIFI_FRAME_TIMEOUT_MS) { disconnectClient(true, DisconnectReason::FrameTimeout); } @@ -42,6 +49,8 @@ void SerialWifiInterface::disconnectClient(bool cancel_session, if (client) client.stop(); resetReceivedFrameHeader(); send_offset = 0; // A reconnect must replay the whole frame, never its suffix. + tx_waiting = false; + tx_last_progress = 0; if (was_connected) { ++disconnect_count; last_disconnect_reason = reason; @@ -243,6 +252,8 @@ const char* SerialWifiInterface::disconnectReasonName(DisconnectReason reason) { case DisconnectReason::Replaced: return "replaced"; case DisconnectReason::Disabled: return "disabled"; case DisconnectReason::Ended: return "ended"; + case DisconnectReason::TxError: return "tx-error"; + case DisconnectReason::TxTimeout: return "tx-timeout"; } return "unknown"; } @@ -264,6 +275,7 @@ size_t SerialWifiInterface::writeFrame(const uint8_t src[], size_t len) { return 0; } send_queue_len = tail_len + pinned; + updateReceiveCapacityPause((uint32_t)millis()); return len; } return 0; @@ -281,11 +293,46 @@ bool SerialWifiInterface::hasReceivedFrameHeader() { return received_frame_header.type != 0 && received_frame_header.length != 0; } +bool SerialWifiInterface::responseCapacityAvailable() const { + if (send_queue_len < FRAME_QUEUE_SIZE) return true; + // A command's reply can displace an unsent best-effort push, never a + // delivery-required reply or the partly transmitted front frame. + for (int i = send_offset != 0 ? 1 : 0; i < send_queue_len; ++i) { + if (!mesh::companionFrameRequiresDelivery(send_queue[i].buf, + send_queue[i].len)) return true; + } + return false; +} + +void SerialWifiInterface::updateReceiveCapacityPause(uint32_t now) { + if (!receive_pending) { + receive_capacity_paused = false; + receive_capacity_paused_at = 0; + return; + } + const bool pause = deviceConnected && !responseCapacityAvailable(); + if (pause && !receive_capacity_paused) { + // A reply-capacity change cannot retroactively forgive a deadline which + // already expired while RX was runnable. + if ((uint32_t)(now - receive_started) >= WIFI_FRAME_TIMEOUT_MS) return; + receive_capacity_paused = true; + receive_capacity_paused_at = now; + } else if (!pause && receive_capacity_paused) { + // Preserve the framing budget already spent, excluding time for which + // this transport intentionally withheld RX to make room for its reply. + receive_started += (uint32_t)(now - receive_capacity_paused_at); + receive_capacity_paused = false; + receive_capacity_paused_at = 0; + } +} + void SerialWifiInterface::resetReceivedFrameHeader() { received_frame_header.type = 0; received_frame_header.length = 0; receive_pending = false; receive_started = 0; + receive_capacity_paused = false; + receive_capacity_paused_at = 0; } size_t SerialWifiInterface::checkRecvFrame(uint8_t dest[]) { @@ -327,23 +374,54 @@ size_t SerialWifiInterface::checkRecvFrame(uint8_t dest[]) { } } - if (deviceConnected) { - if (send_queue_len > 0) { // first, check send queue - - _last_write = millis(); - int len = send_queue[0].len; + if (deviceConnected && send_queue_len > 0) { + _last_write = millis(); + const uint32_t tx_now = (uint32_t)_last_write; + if (!tx_waiting) { + tx_waiting = true; + tx_last_progress = tx_now; + } + int len = send_queue[0].len; - 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); - 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; + 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); + const size_t total = 3U + len; + const size_t remaining = total - send_offset; + size_t written = 0; +#if defined(ESP32) + // WiFiClient::write retries ten one-second select waits and restarts that + // budget after partial progress. One nonblocking send bounds this poll + // and leaves both incoming commands and other transports runnable. + const int result = ::send(client.fd(), pkt + send_offset, remaining, + MSG_DONTWAIT); + if (result > 0) { + written = static_cast(result); + } else if (result < 0 && errno != EAGAIN && errno != EWOULDBLOCK + && errno != EINTR) { + disconnectClient(true, DisconnectReason::TxError); + return 0; + } +#else + written = client.write(pkt + send_offset, remaining); +#endif + if (written > 0) { + tx_last_progress = (uint32_t)millis(); + } else if ((uint32_t)((uint32_t)millis() - tx_last_progress) + >= WIFI_TX_STALL_TIMEOUT_MS) { + // A locked stream on another transport may omit WiFi's polling for a + // while. Try its socket before declaring a stall: a writable client + // can resume after that scheduler gap without an artificial timeout. + disconnectClient(true, DisconnectReason::TxTimeout); + return 0; + } + send_offset += written < remaining ? written : remaining; + if (send_offset == total) { send_offset = 0; + tx_waiting = false; + tx_last_progress = 0; ++tx_frames; last_tx_opcode = send_queue[0].buf[0]; send_queue_len--; @@ -351,42 +429,46 @@ size_t SerialWifiInterface::checkRecvFrame(uint8_t dest[]) { send_queue[i] = send_queue[i + 1]; } memset(&send_queue[send_queue_len], 0, sizeof(send_queue[send_queue_len])); - } else { - - if (!receive_pending && client.available() > 0) { - receive_pending = true; - receive_started = (uint32_t)millis(); - } - if (!hasReceivedFrameHeader()) { - if (client.available() < 3) return 0; - uint8_t header[3]; - if (client.readBytes(header, sizeof(header)) != sizeof(header)) { - disconnectClient(true, DisconnectReason::ShortHeader); - return 0; - } - received_frame_header.type = header[0]; - received_frame_header.length = uint16_t(header[1]) | (uint16_t(header[2]) << 8); - // Validate before waiting for any payload. Closing an invalid stream - // avoids unbounded discard loops and interpreting its tail as commands. - if (received_frame_header.type != '<' || received_frame_header.length == 0 - || received_frame_header.length > MAX_FRAME_SIZE) { - disconnectClient(true, DisconnectReason::InvalidHeader); - return 0; - } - } - const size_t frame_length = received_frame_header.length; - if (client.available() < (int)frame_length) return 0; - if (dest == nullptr || client.readBytes(dest, frame_length) != frame_length) { - disconnectClient(true, DisconnectReason::ShortPayload); - return 0; // Never dispatch a truncated command, even after a short read. - } - resetReceivedFrameHeader(); - ++valid_frames; - last_opcode = dest[0]; - return frame_length; - } } + // Sending a queued push or just part of a frame must not postpone the + // client's device-query/app-start indefinitely. Poll at most one incoming + // frame independently of the single outgoing attempt above. + updateReceiveCapacityPause((uint32_t)millis()); + if (deviceConnected) { + if (!responseCapacityAvailable()) return 0; + if (!receive_pending && client.available() > 0) { + receive_pending = true; + receive_started = (uint32_t)millis(); + } + if (!hasReceivedFrameHeader()) { + if (client.available() < 3) return 0; + uint8_t header[3]; + if (client.readBytes(header, sizeof(header)) != sizeof(header)) { + disconnectClient(true, DisconnectReason::ShortHeader); + return 0; + } + received_frame_header.type = header[0]; + received_frame_header.length = uint16_t(header[1]) | (uint16_t(header[2]) << 8); + // Validate before waiting for any payload. Closing an invalid stream + // avoids unbounded discard loops and interpreting its tail as commands. + if (received_frame_header.type != '<' || received_frame_header.length == 0 + || received_frame_header.length > MAX_FRAME_SIZE) { + disconnectClient(true, DisconnectReason::InvalidHeader); + return 0; + } + } + const size_t frame_length = received_frame_header.length; + if (client.available() < (int)frame_length) return 0; + if (dest == nullptr || client.readBytes(dest, frame_length) != frame_length) { + disconnectClient(true, DisconnectReason::ShortPayload); + return 0; // Never dispatch a truncated command, even after a short read. + } + resetReceivedFrameHeader(); + ++valid_frames; + last_opcode = dest[0]; + return frame_length; + } return 0; } diff --git a/src/helpers/wifi/SerialWifiInterface.h b/src/helpers/wifi/SerialWifiInterface.h index 9b03b1f7..f54b84ed 100644 --- a/src/helpers/wifi/SerialWifiInterface.h +++ b/src/helpers/wifi/SerialWifiInterface.h @@ -10,7 +10,7 @@ public: enum class StartError : uint8_t { None, InvalidPort, StartFailed }; enum class DisconnectReason : uint8_t { None, PeerClosed, FrameTimeout, InvalidHeader, ShortHeader, ShortPayload, - Replaced, Disabled, Ended + Replaced, Disabled, Ended, TxError, TxTimeout }; struct Status { State state; @@ -60,6 +60,8 @@ private: uint8_t last_opcode = 0; uint32_t tx_frames = 0; uint8_t last_tx_opcode = 0; + bool tx_waiting = false; + uint32_t tx_last_progress = 0; uint32_t disconnect_count = 0; DisconnectReason last_disconnect_reason = DisconnectReason::None; @@ -89,6 +91,8 @@ private: size_t send_offset = 0; uint32_t receive_started = 0; bool receive_pending = false; + bool receive_capacity_paused = false; + uint32_t receive_capacity_paused_at = 0; void (*session_changed)(void*) = nullptr; void* session_context = nullptr; @@ -99,6 +103,8 @@ private: bool addressReady(); bool serviceListener(uint32_t now); void stopListener(DisconnectReason reason); + bool responseCapacityAvailable() const; + void updateReceiveCapacityPause(uint32_t now); protected: diff --git a/test/fixtures/companion_wifi_listener/test.cpp b/test/fixtures/companion_wifi_listener/test.cpp index 0e66b28a..9a215c1c 100644 --- a/test/fixtures/companion_wifi_listener/test.cpp +++ b/test/fixtures/companion_wifi_listener/test.cpp @@ -1,5 +1,6 @@ #include #include +#include #include #include #include @@ -14,13 +15,27 @@ using Bytes = std::vector; static const IPAddress A(192, 168, 1, 20); static const IPAddress B(192, 168, 1, 30); +static Bytes framed(std::initializer_list frames) { + Bytes result; + for (const auto& frame : frames) { + result.insert(result.end(), {'>', uint8_t(frame.size()), 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 cancellations = 0; + MultiSerialInterface* manager = nullptr; + bool pending_operation = false; Fixture() { transport.setSessionChangedCallback([](void* context) { - ++static_cast(context)->cancellations; + auto& f = *static_cast(context); + ++f.cancellations; + f.pending_operation = false; + if (f.manager) f.manager->forgetReplyRouteForDisconnected(&f.transport); }, this); } ~Fixture() { transport.end(); } @@ -33,6 +48,13 @@ struct Fixture { tick(); return socket; } + void enqueue(const Bytes& data) { + assert(transport.writeFrame(data.data(), data.size()) == data.size()); + } + void drain(const Socket& socket) { + socket->write_limit = std::numeric_limits::max(); + for (int i = 0; i < 20; ++i) tick(); + } }; int main(int argc, char** argv) { @@ -341,6 +363,244 @@ int main(int argc, char** argv) { assert(!f.start()); f.tick(); assert(!f.transport.getStatus().addressReady && WiFiServer::begin_calls == 0); + } else if (scenario == 28 || scenario == 31 || scenario == 32 || scenario == 33) { + assert(f.start()); + auto socket = f.connect(A); + f.enqueue(Bytes{0x80, 0xa1}); + if (scenario == 28) socket->write_limit = std::numeric_limits::max(); + if (scenario == 32) socket->send_error = EAGAIN; + if (scenario == 33) socket->send_error = EINTR; + if (scenario == 31) { + socket->sdk_write_block_ms = 10000; + MockTcp::sdk_write_hook = [](uint32_t delay) { mock_millis += delay; }; + } + socket->received = {'<', 2, 0, 0x16, 3}; + const auto before = mock_millis; + assert(f.tick() == 2 && f.input[0] == 0x16 && f.input[1] == 3); + assert(mock_millis == before && socket->sdk_write_calls == 0 && socket->send_calls == 1); + assert(socket->last_send_flags & MSG_DONTWAIT); + assert(socket->connected && f.cancellations == 0); + if (scenario != 28) { + assert(socket->sent.empty() && f.transport.hasPendingIO()); + socket->send_error = 0; + f.drain(socket); + assert((socket->sent == framed({Bytes{0x80, 0xa1}}))); + } + } else if (scenario == 29) { + assert(f.start()); + auto socket = f.connect(A); + for (int poll = 0; poll < 100; ++poll) { + const Bytes push{0x88, uint8_t(poll)}; + // Production permits shedding best-effort radio logs when saturated. + f.transport.writeFrame(push.data(), push.size()); + socket->received = {'<', 2, 0, 0x16, 3}; + assert(f.tick() == 2 && f.input[0] == 0x16 && f.input[1] == 3); + ++mock_millis; + } + auto status = f.transport.getStatus(); + assert(status.validFrames == 100 && status.acceptedClients == 1); + assert(socket->sdk_write_calls == 0 && socket->send_calls == 100); + assert(status.clientConnected && f.cancellations == 0 && f.transport.hasPendingIO()); + } else if (scenario == 30) { + assert(f.start()); + auto socket = f.connect(A); + const Bytes first{0x80, 0xa1}, second{0x88, 0xb1}, info{0x0d, 0x31, 0x32}; + f.enqueue(first); f.enqueue(second); + socket->write_limit = 1; + socket->received = {'<', 2, 0, 0x16, 3}; + assert(f.tick() == 2 && socket->sent == Bytes{'>'}); + assert(f.transport.getStatus().txFrames == 0); + f.enqueue(info); + f.drain(socket); + assert(socket->sent == framed({first, info, second})); + assert(f.transport.getStatus().txFrames == 3 && f.transport.getStatus().validFrames == 1); + } else if (scenario == 34 || scenario == 35 || scenario == 42) { + assert(f.start()); + auto socket = f.connect(A); + MultiSerialInterface manager; + f.manager = &manager; + assert(manager.addInterface(InterfaceType::WiFi, &f.transport)); manager.enable(); + socket->received = {'<', 2, 0, 0x16, 3}; + assert(manager.checkRecvFrame(f.input) == 2); manager.lockReplyRoute(); + f.pending_operation = true; + const Bytes reply{0x0d, 0x31, 0x32}; + assert(manager.writeFrame(reply.data(), reply.size()) == reply.size()); + socket->write_limit = 2; f.tick(); + assert(socket->sent.size() == 2); + if (scenario == 34 || scenario == 42) socket->send_error = EPIPE; + else socket->descriptor_override = -1; + socket->received = {'<', 2, 0, 0x16, 3}; + assert(f.tick() == 0); + assert(!socket->connected && f.cancellations == 1 && !f.pending_operation); + assert(!manager.isReplyRouteAvailable(&f.transport)); + assert(f.transport.getStatus().lastDisconnectReason == SerialWifiInterface::DisconnectReason::TxError); + assert(socket->read_count == 5); // Fatal TX must not execute the next command. + f.transport.loop(); assert(f.cancellations == 1); + f.manager = nullptr; + if (scenario == 42) { + auto other = f.connect(B, Bytes{'<', 2, 0, 0x16, 3}); + const Bytes other_reply{0x0d, 0x41}; f.enqueue(other_reply); f.drain(other); + assert(other->sent == framed({other_reply})); + } + auto returned = f.connect(A); f.drain(returned); + assert(returned->sent == framed({reply})); + } else if (scenario == 36 || scenario == 37) { + assert(f.start()); + auto socket = f.connect(A); + if (scenario == 37) mock_millis = UINT32_MAX - 5000; + const Bytes reply{0x0d, 0x51}; f.enqueue(reply); f.tick(); + mock_millis += 9999; f.tick(); + assert(socket->connected && f.cancellations == 0); + ++mock_millis; f.tick(); + assert(!socket->connected && f.cancellations == 1); + assert(f.transport.getStatus().lastDisconnectReason == SerialWifiInterface::DisconnectReason::TxTimeout); + auto returned = f.connect(A); f.drain(returned); + assert(returned->sent == framed({reply})); + } else if (scenario == 38) { + assert(f.start()); + auto socket = f.connect(A); f.enqueue(Bytes{0x0d, 0x51}); f.tick(); + mock_millis += 9999; socket->write_limit = 1; f.tick(); + assert(socket->connected && socket->sent.size() == 1); + socket->write_limit = 0; + mock_millis += 9999; f.tick(); + assert(socket->connected && f.cancellations == 0); + ++mock_millis; f.tick(); + assert(!socket->connected && f.cancellations == 1); + assert(f.transport.getStatus().lastDisconnectReason == SerialWifiInterface::DisconnectReason::TxTimeout); + } else if (scenario == 39) { + assert(f.start()); + auto socket = f.connect(A); f.enqueue(Bytes{0x0d, 0x51}); f.drain(socket); + mock_millis += 60000; f.transport.loop(); + assert(socket->connected && f.cancellations == 0); + socket->write_limit = 0; f.enqueue(Bytes{0x01, 0x61}); f.tick(); + mock_millis += 9999; f.tick(); + assert(socket->connected); + ++mock_millis; f.tick(); + assert(!socket->connected && f.cancellations == 1); + } else if (scenario == 40) { + assert(f.start()); + auto socket = f.connect(A); f.enqueue(Bytes{0x88, 0x51}); + socket->received = {'<', 2}; f.tick(); + mock_millis += 4999; f.tick(); + assert(socket->connected); + ++mock_millis; f.transport.loop(); + assert(!socket->connected && f.cancellations == 1); + assert(f.transport.getStatus().lastDisconnectReason == SerialWifiInterface::DisconnectReason::FrameTimeout); + } else if (scenario == 41) { + assert(f.start()); + auto old = f.connect(A); const Bytes reply{0x0d, 0x51}; f.enqueue(reply); f.tick(); + mock_millis += 9999; + auto next = f.connect(A); + assert(!old->connected && next->connected && f.cancellations == 0); + mock_millis += 9999; f.transport.loop(); + assert(next->connected && f.cancellations == 0); + f.drain(next); assert(next->sent == framed({reply})); + } else if (scenario == 43) { + assert(f.start()); + auto socket = f.connect(A); f.enqueue(Bytes{0x0d, 0x51}); + socket->send_error = EWOULDBLOCK; socket->received = {'<', 2, 0, 0x16, 3}; + assert(f.tick() == 2 && socket->connected && f.cancellations == 0); + socket->send_error = ECONNRESET; assert(f.tick() == 0); + assert(!socket->connected && f.cancellations == 1); + assert(f.transport.getStatus().lastDisconnectReason == SerialWifiInterface::DisconnectReason::TxError); + } else if (scenario == 44) { + assert(f.start()); + auto socket = f.connect(A); + for (int i = 0; i < 4; ++i) f.enqueue(Bytes{0x01, uint8_t(i)}); + socket->received = {'<', 2, 0, 0x16, 3}; + assert(f.tick() == 0 && socket->read_count == 0 && f.transport.getStatus().validFrames == 0); + socket->write_limit = 2; assert(f.tick() == 0 && socket->read_count == 0); + socket->write_limit = std::numeric_limits::max(); + assert(f.tick() == 2 && f.input[0] == 0x16 && f.input[1] == 3); + f.enqueue(Bytes{0x0d, 0x51}); + f.drain(socket); + assert((socket->sent == framed({Bytes{0x01, 0}, Bytes{0x01, 1}, Bytes{0x01, 2}, Bytes{0x01, 3}, Bytes{0x0d, 0x51}}))); + } else if (scenario == 45) { + assert(f.start()); + auto socket = f.connect(A); + const Bytes best{0x88, 0x41}; f.enqueue(best); + for (int i = 0; i < 3; ++i) f.enqueue(Bytes{0x01, uint8_t(i)}); + socket->received = {'<', 2, 0, 0x16, 3}; + assert(f.tick() == 2); + f.enqueue(Bytes{0x0d, 0x51}); f.drain(socket); + assert((socket->sent == framed({Bytes{0x01, 0}, Bytes{0x01, 1}, Bytes{0x01, 2}, Bytes{0x0d, 0x51}}))); + } else if (scenario == 46) { + assert(f.start()); + auto socket = f.connect(A); const Bytes best{0x88, 0x41}; f.enqueue(best); + socket->write_limit = 1; f.tick(); + assert(socket->sent == Bytes{'>'}); + for (int i = 0; i < 3; ++i) f.enqueue(Bytes{0x01, uint8_t(i)}); + socket->write_limit = 0; socket->received = {'<', 2, 0, 0x16, 3}; + assert(f.tick() == 0 && socket->read_count == 0); + socket->write_limit = std::numeric_limits::max(); + assert(f.tick() == 2); f.enqueue(Bytes{0x0d, 0x51}); f.drain(socket); + assert((socket->sent == framed({best, Bytes{0x01, 0}, Bytes{0x01, 1}, Bytes{0x01, 2}, Bytes{0x0d, 0x51}}))); + } else if (scenario == 47) { + assert(f.start()); + auto socket = f.connect(A, Bytes{'<', 2, 0, 0x16}); + assert(socket->read_count == 3 && f.transport.hasReceivedFrameHeader()); + for (int i = 0; i < 4; ++i) f.enqueue(Bytes{0x01, uint8_t(i)}); + socket->received.push_back(3); assert(f.tick() == 0); + mock_millis += 6000; f.transport.loop(); + assert(socket->connected && f.cancellations == 0 && f.transport.hasReceivedFrameHeader()); + socket->write_limit = std::numeric_limits::max(); + assert(f.tick() == 2 && f.input[0] == 0x16 && f.input[1] == 3); + assert(socket->connected && f.cancellations == 0); + } else if (scenario == 48) { + assert(f.start()); + auto socket = f.connect(A); const Bytes reply{0x0d, 0x51, 0x52}; + f.enqueue(reply); socket->write_limit = 1; f.tick(); + struct HoldingInterface : BaseSerialInterface { + bool input = true; + void enable() override {} + void disable() override {} + bool isEnabled() const override { return true; } + bool isConnected() const override { return true; } + bool isReadBusy() const override { return false; } + bool isWriteBusy() const override { return false; } + size_t writeFrame(const uint8_t[], size_t size) override { return size; } + size_t checkRecvFrame(uint8_t dest[]) override { + if (!input) return 0; + input = false; dest[0] = 0x04; return 1; + } + } other; + MultiSerialInterface manager; + assert(manager.addInterface(InterfaceType::WiFi, &f.transport)); + assert(manager.addInterface(InterfaceType::USB, &other)); manager.enable(); + assert(manager.checkRecvFrame(f.input) == 1 && f.input[0] == 0x04); + manager.lockReplyRoute(); + const size_t sent = socket->sent.size(), sends = socket->send_calls; + for (int i = 0; i < 20; ++i) { + mock_millis += 1000; manager.loop(); + assert(manager.checkRecvFrame(f.input) == 0); + } + assert(socket->connected && f.cancellations == 0); + assert(socket->sent.size() == sent && socket->send_calls == sends); + manager.unlockReplyRoute(); socket->write_limit = std::numeric_limits::max(); + manager.checkRecvFrame(f.input); + assert(socket->connected && f.cancellations == 0 && socket->sent == framed({reply})); + } else if (scenario == 49 || scenario == 51) { + assert(f.start()); + if (scenario == 51) mock_millis = UINT32_MAX - 3000; + auto socket = f.connect(A, Bytes{'<', 2, 0, 0x16}); + mock_millis += 4000; + for (int i = 0; i < 4; ++i) f.enqueue(Bytes{0x01, uint8_t(i)}); + f.tick(); mock_millis += 6000; f.transport.loop(); + assert(socket->connected && f.cancellations == 0 && f.transport.hasReceivedFrameHeader()); + socket->write_limit = std::numeric_limits::max(); f.tick(); // Frees a reply slot, but body is incomplete. + mock_millis += 999; f.tick(); + assert(socket->connected && f.cancellations == 0); + ++mock_millis; f.transport.loop(); + assert(!socket->connected && f.cancellations == 1); + assert(f.transport.getStatus().lastDisconnectReason == SerialWifiInterface::DisconnectReason::FrameTimeout); + } else if (scenario == 50) { + assert(f.start()); + auto socket = f.connect(A, Bytes{'<', 2, 0, 0x16}); + mock_millis += 5000; + for (int i = 0; i < 4; ++i) f.enqueue(Bytes{0x01, uint8_t(i)}); + f.transport.loop(); + assert(!socket->connected && f.cancellations == 1); + assert(f.transport.getStatus().lastDisconnectReason == SerialWifiInterface::DisconnectReason::FrameTimeout); } else { assert(false && "unknown scenario"); } diff --git a/test/fixtures/serial_wifi_sessions/mocks/WiFi.h b/test/fixtures/serial_wifi_sessions/mocks/WiFi.h index 988ee61d..f601c189 100644 --- a/test/fixtures/serial_wifi_sessions/mocks/WiFi.h +++ b/test/fixtures/serial_wifi_sessions/mocks/WiFi.h @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -48,10 +49,25 @@ struct WiFiClass { inline WiFiClass WiFi; +struct MockSocket; +namespace MockTcp { +inline int next_descriptor = 3; +inline std::map> sockets; +inline void (*sdk_write_hook)(uint32_t) = nullptr; +} + struct MockSocket { IPAddress address; bool connected = true; + int descriptor = MockTcp::next_descriptor++; + int descriptor_override = -2; size_t write_limit = 0; + size_t sdk_write_calls = 0; + uint32_t sdk_write_block_ms = 0; + size_t send_calls = 0; + int last_send_flags = 0; + int send_error = 0; + std::deque send_results; size_t read_limit = std::numeric_limits::max(); size_t read_count = 0; std::vector sent; @@ -63,11 +79,18 @@ class WiFiClient { std::shared_ptr _socket; public: WiFiClient() = default; - explicit WiFiClient(std::shared_ptr socket) : _socket(socket) {} + explicit WiFiClient(std::shared_ptr socket) : _socket(socket) { + MockTcp::sockets[_socket->descriptor] = _socket; + } explicit operator bool() { return _socket && _socket->connected; } // Match the pinned ESP32 SDK's non-const connected() signature. This catches // const transport accessors that accidentally rely on a const-only fake. bool connected() { return _socket && _socket->connected; } + // Public SDK 2.x/3.x API; no test-only access to a protected socket handle. + virtual int fd() const { + if (!_socket) return -1; + return _socket->descriptor_override == -2 ? _socket->descriptor : _socket->descriptor_override; + } void stop() { if (_socket) _socket->connected = false; } IPAddress remoteIP() const { return _socket ? _socket->address : IPAddress(); } int available() { return connected() ? int(_socket->received.size()) : 0; } @@ -84,6 +107,8 @@ public: 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; + ++_socket->sdk_write_calls; + if (MockTcp::sdk_write_hook) MockTcp::sdk_write_hook(_socket->sdk_write_block_ms); const size_t written = std::min(size, _socket->write_limit); _socket->sent.insert(_socket->sent.end(), src, src + written); return written; @@ -108,6 +133,9 @@ public: listener_lost = false; incoming.clear(); WiFi = WiFiClass{}; + MockTcp::sockets.clear(); + MockTcp::next_descriptor = 3; + MockTcp::sdk_write_hook = nullptr; } void begin(int port) { ++begin_calls; diff --git a/test/fixtures/serial_wifi_sessions/mocks/lwip/sockets.h b/test/fixtures/serial_wifi_sessions/mocks/lwip/sockets.h new file mode 100644 index 00000000..277e7d40 --- /dev/null +++ b/test/fixtures/serial_wifi_sessions/mocks/lwip/sockets.h @@ -0,0 +1,37 @@ +#pragma once + +#include +#include +#include + +#ifndef MSG_DONTWAIT +#define MSG_DONTWAIT 0x40 +#endif + +// One bounded socket send. Unlike SDK WiFiClient::write(), this never polls or +// advances the mock clock. Scripted negative results inject errno, not bytes. +inline int send(int descriptor, const void* source, size_t size, int flags) { + const auto found = MockTcp::sockets.find(descriptor); + if (found == MockTcp::sockets.end()) { errno = EBADF; return -1; } + const auto socket = found->second.lock(); + if (!socket || !socket->connected) { errno = EPIPE; return -1; } + ++socket->send_calls; + socket->last_send_flags = flags; + assert(flags & MSG_DONTWAIT); + int result = 0; + if (!socket->send_results.empty()) { + result = socket->send_results.front(); + socket->send_results.pop_front(); + if (result < 0) { errno = -result; return -1; } + } else if (socket->send_error) { + errno = socket->send_error; + return -1; + } else { + result = static_cast(std::min(size, socket->write_limit)); + } + const size_t written = std::min(size, static_cast(result)); + const auto* bytes = static_cast(source); + socket->sent.insert(socket->sent.end(), bytes, bytes + written); + errno = 0; + return static_cast(written); +} diff --git a/test/test_companion_wifi_listener.py b/test/test_companion_wifi_listener.py index 45bd66e1..97d0b4ce 100644 --- a/test/test_companion_wifi_listener.py +++ b/test/test_companion_wifi_listener.py @@ -3,6 +3,7 @@ from pathlib import Path import os +import re import shutil import subprocess import sys @@ -48,6 +49,30 @@ SCENARIOS = ( "MultiSerial services unavailable transport so it can recover before dispatch", "failed listener repair retains an accepted client and queued data", "stale AP address never starts a listener while AP mode is disabled", + "queued radio push cannot hide an already complete DEVICE_QUERY", + "continuous unsolicited pushes cannot starve 100 incoming handshakes", + "partial TX cannot block RX or let a new response corrupt its pinned frame", + "zero TX never calls the blocking SDK writer or blocks incoming DEVICE_QUERY", + "EAGAIN retains TX and still dispatches complete incoming commands", + "EINTR retains TX and still dispatches complete incoming commands", + "fatal EPIPE cancels the route once and replays the complete frame on reconnect", + "invalid public socket descriptor cancels the route without consuming input", + "TX no-progress deadline fires at ten seconds and permits same-IP queue replay", + "TX no-progress deadline survives millis rollover", + "positive byte progress resets the TX deadline but zero writes do not", + "completed TX and a later new frame have independent stall deadlines", + "partial inbound frame still expires when outbound traffic is stalled", + "same-IP live takeover resets stalled TX and preserves complete replies", + "peer takeover after TX error receives none of the old client replies", + "ECONNRESET is fatal while EWOULDBLOCK remains temporary", + "full required queue defers incoming command until its reply can be admitted", + "unpinned best-effort queue item permits a higher priority handshake reply", + "only pinned best-effort frame cannot be evicted to admit a handshake reply", + "deliberate reply-capacity wait does not expire a partial inbound frame", + "another transport lock cannot cause a false timeout on a writable WiFi client", + "reply-capacity pause preserves the remaining inbound timeout budget", + "new queue pressure cannot retroactively forgive an expired inbound frame", + "reply-capacity timeout pause survives millis rollover", ) @@ -75,6 +100,40 @@ class CompanionWiFiListenerTests(unittest.TestCase): self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr) self.assertIn(f"PASS: Wi-Fi listener scenario {scenario}", checked.stdout) + def test_actual_device_query_response_survives_continuous_radio_pushes(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") + mesh = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text(encoding="utf-8") + header = (ROOT / "examples/companion_radio/MyMesh.h").read_text(encoding="utf-8") + query = extract_braced(mesh, "if (cmd_frame[0] == CMD_DEVICE_QUERY && len >= 2)") + helpers = (ROOT / "src/helpers/TxtDataHelpers.cpp").read_text(encoding="utf-8") + copy = extract_braced(helpers, "void StrHelper::strzcpy(") + constants = [] + for name, source in (("CMD_DEVICE_QUERY", mesh), ("RESP_CODE_DEVICE_INFO", mesh), + ("FIRMWARE_VER_CODE", header)): + match = re.search(r"^#define\s+" + name + r"\s+(\d+)\b", source, re.MULTILINE) + self.assertIsNotNone(match, name) + constants.append(f"#define {name} {match.group(1)}") + fixture = (ROOT / "test/fixtures/companion_wifi_listener/handshake.cpp").read_text() + fixture = fixture.replace("@CONSTANTS@", "\n".join(constants)) + fixture = fixture.replace("@DEVICE_QUERY@", query).replace("@COPY@", copy) + with tempfile.TemporaryDirectory(prefix="meshcore-wifi-handshake-") as directory: + work = Path(directory) + source = work / "handshake.cpp" + source.write_text(fixture, encoding="utf-8") + binary = work / "handshake.exe" + compiled = subprocess.run( + [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", + *SANITIZERS, "-DESP32=1", f"-I{MOCKS}", f"-I{ROOT / 'src'}", + str(source), str(ROOT / "src/helpers/wifi/SerialWifiInterface.cpp"), + "-o", str(binary)], capture_output=True, text=True, timeout=60, + ) + self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr) + checked = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10) + self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr) + self.assertIn("PASS: actual DEVICE_QUERY reply under 100 radio-push polls", checked.stdout) + def test_production_cli_and_display_formatter_are_bounded_and_observable(self): compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++") if compiler is None: diff --git a/test/test_serial_wifi_sessions.py b/test/test_serial_wifi_sessions.py index 77a5db54..89060906 100644 --- a/test/test_serial_wifi_sessions.py +++ b/test/test_serial_wifi_sessions.py @@ -65,25 +65,26 @@ class SerialWifiSessionTests(unittest.TestCase): 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/wifi/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) + for platform, flags in (("generic", []), ("esp32", ["-DESP32=1"])): + binary = Path(directory) / f"wifi-sessions-{platform}.exe" + compiled = subprocess.run( + [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", + *SANITIZER_FLAGS, *flags, + f"-I{FIXTURE / 'mocks'}", f"-I{ROOT / 'src'}", + str(FIXTURE / "test.cpp"), + str(ROOT / "src/helpers/wifi/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(platform=platform, 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__":