From 2e84512ba47549552430b95ae23225fde7342aa9 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Sat, 3 Oct 2026 23:22:34 -0700 Subject: [PATCH] Retain and isolate delayed Companion results under backpressure --- .github/workflows/run-unit-tests.yml | 1 + docs/companion_radio_full.md | 29 + examples/companion_radio/MyMesh.cpp | 428 +++++----- examples/companion_radio/MyMesh.h | 31 +- src/helpers/BaseChatMesh.cpp | 3 + src/helpers/BaseChatMesh.h | 2 + src/helpers/CompanionDelayedReplies.cpp | 303 +++++++ src/helpers/CompanionDelayedReplies.h | 93 +++ test/fixtures/companion_ack_delivery/test.cpp | 10 +- .../companion_delayed_reply_delivery/test.cpp | 760 ++++++++++++++++++ .../contact_cache/test_contact_cache.cpp | 1 + test/test_companion_ack_delivery.py | 1 + test/test_companion_delayed_reply_delivery.py | 143 ++++ test/test_companion_discovery_correlation.py | 89 +- ...st_companion_radio_settings_transaction.py | 29 +- test/test_companion_response_bounds.py | 68 +- test/test_companion_wifi_session.py | 28 +- test/test_esp32_usb_serial_hygiene.py | 10 +- test/test_nrf52_usb_logging_contract.py | 59 +- test/test_usb_logging_watchdog_roles.py | 50 +- 20 files changed, 1798 insertions(+), 340 deletions(-) create mode 100644 src/helpers/CompanionDelayedReplies.cpp create mode 100644 src/helpers/CompanionDelayedReplies.h create mode 100644 test/fixtures/companion_delayed_reply_delivery/test.cpp create mode 100644 test/test_companion_delayed_reply_delivery.py diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index 6331f4b21..b3ab6b759 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -134,6 +134,7 @@ jobs: python3 -B test/test_companion_primary_radio_persistence.py -v python3 -B test/test_companion_response_bounds.py -v python3 -B test/test_companion_ack_delivery.py -v + python3 -B test/test_companion_delayed_reply_delivery.py -v python3 -B test/test_companion_hardware_serial.py -v python3 -B test/test_companion_hardware_serial_gps.py -v python3 -B test/test_companion_contact_lifetime.py -v diff --git a/docs/companion_radio_full.md b/docs/companion_radio_full.md index 06d142224..04b8d970b 100644 --- a/docs/companion_radio_full.md +++ b/docs/companion_radio_full.md @@ -558,6 +558,35 @@ 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 later frame can interleave with it or cause it to be discarded. +Binary login, status, telemetry, path-discovery, binary-request, and trace +results also have bounded producer retention when that transport queue is +full. One request result and one trace result can wait independently, on their +original captured interfaces. Their required `SENT` frames are retained and +admitted before the final result. A first valid radio result gets up to ten +seconds for whole-frame queue admission; a still-blocked `SENT` keeps its +earlier ten-second deadline. Duplicate callbacks cannot replace the retained +result or extend these deadlines. Disconnect, cancellation, or expiry drops +an unadmitted result. Queue admission is not proof of physical host receipt. + +Reflected request tags are matched against the complete peer public key. +Recent request and trace history has eight slots each, reserved before radio +transmission; live entries are never evicted to accept a newer request. A full +history or a quarantined identifier returns `ERR_BAD_STATE`, so clients must +back off. Retired reflected tags and trace tag/auth tuples are suppressed for +at least ten seconds and through the original radio deadline plus ten seconds +when longer. Login retries to the same peer have at least a thirty-second +quarantine. Positively matched newer status/telemetry/binary replies still work +after login, but otherwise unsolicited responses from that login peer can be +suppressed during the quarantine. + +These guards are finite and clear at reboot. Login replies carry a server +timestamp, not an echoed request nonce; sufficiently late replies or a +timestamp collision cannot always be distinguished without a protocol change. +Trace tag/auth fields are correlation data, not cryptographic authentication. +The guards do not authenticate a host process or retract frames already +accepted by a transport. In particular, the same-IP TCP queue replay described +above remains unchanged, including its lack of authentication. + When a BLE client requests pairing, a display-equipped build wakes the screen, switches to the first home page, and keeps the active six-digit PIN visible until Bluetooth connects or the two-minute pairing window expires. USB, WiFi, diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 6fae1445e..cba311789 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -357,9 +357,10 @@ void MyMesh::writeErrFrame(uint8_t err_code, BaseSerialInterface* route) { } } -size_t MyMesh::writePendingSerialFrame(const uint8_t frame[], size_t len) { - if (_serial == NULL || pending_serial_reply_route == NULL) return 0; - return _serial->writeFrameToRoute(pending_serial_reply_route, frame, len); +size_t MyMesh::writePendingSerialFrame(const uint8_t frame[], size_t len, uint32_t now) { + if (!_delayed_replies.storeRequest(frame, len, now)) return 0; + _delayed_replies.serviceRequest(_serial, now); + return len; // Retained by the producer, not necessarily admitted yet. } void MyMesh::writeDisabledFrame() { @@ -1935,6 +1936,8 @@ uint8_t MyMesh::onContactRequest(const ContactInfo &contact, uint32_t sender_tim } void MyMesh::onContactResponse(const ContactInfo &contact, const uint8_t *data, uint8_t len) { + const uint32_t now = _ms->getMillis(); + servicePendingSerialReply(now); // Expire before considering any radio callback. if (data == NULL || len < 4) return; #if MESH_ENABLE_ONE_KEY_DM @@ -1961,10 +1964,21 @@ void MyMesh::onContactResponse(const ContactInfo &contact, const uint8_t *data, uint32_t tag; memcpy(&tag, data, 4); - if (pending_login && memcmp(&pending_login, contact.id.pub_key, 4) == 0) { // check for login response + const bool matched = _delayed_replies.requestMatches(contact.id.pub_key, tag); + const auto kind = _delayed_replies.request.kind; + if (matched && _delayed_replies.request.phase != mesh::CompanionDelayedReplies::AwaitRadio) return; + // Login has no reflected nonce. Known old request replies take precedence + // over its permissive envelope; exact newer reflected replies remain valid + // immediately after a successful login despite the login quarantine. + if (matched && kind == mesh::CompanionDelayedReplies::Login + && _delayed_replies.blocksResponse(contact.id.pub_key, tag, now)) return; + + if (matched && kind == mesh::CompanionDelayedReplies::Login) { + if (len < 6 || (data[4] == RESP_SERVER_LOGIN_OK && len < 13)) return; // yes, is response to pending sendLogin() + uint16_t keep_alive_secs = 0; #if COMPANION_FEATURE_TEXT_TERMINAL - const bool terminal_login_response = _terminal_login_pending + const bool terminal_login_response = _delayed_replies.request.terminal && _terminal_login_pending && memcmp(_terminal_login_key, contact.id.pub_key, sizeof(_terminal_login_key)) == 0; #endif @@ -1986,10 +2000,7 @@ void MyMesh::onContactResponse(const ContactInfo &contact, const uint8_t *data, login_success = true; modern_login = true; #endif - uint16_t keep_alive_secs = ((uint16_t)data[5]) * 16; - if (keep_alive_secs > 0) { - startConnection(contact, keep_alive_secs); - } + keep_alive_secs = ((uint16_t)data[5]) * 16; out_frame[i++] = PUSH_CODE_LOGIN_SUCCESS; out_frame[i++] = data[6]; // permissions (eg. is_admin) memcpy(&out_frame[i], contact.id.pub_key, 6); @@ -2007,7 +2018,9 @@ void MyMesh::onContactResponse(const ContactInfo &contact, const uint8_t *data, // Terminal login has its own text destination and deliberately captures // no Binary route. Binary logins retain their exact requester even if a // different transport has issued a command in the meantime. - writePendingSerialFrame(out_frame, i); + _delayed_replies.rememberLoginTag(tag); + if (writePendingSerialFrame(out_frame, i, now) != (size_t)i) return; + if (keep_alive_secs > 0) startConnection(contact, keep_alive_secs); #if COMPANION_FEATURE_TEXT_TERMINAL if (terminal_login_response) { if (hasTerminalOutput()) { @@ -2024,11 +2037,10 @@ void MyMesh::onContactResponse(const ContactInfo &contact, const uint8_t *data, _terminal_login_target); } } - clearTerminalLogin(); + clearTerminalLogin(now); } #endif - clearPendingReqs(); - } else if (mesh::companionStatusTagMatches(pending_status, tag)) { + } else if (matched && kind == mesh::CompanionDelayedReplies::Status) { // Require a complete response that also fits its host-protocol envelope. // The app parses at least 48 bytes and otherwise throws a RangeError. if (!mesh::companionStatusResponseIsLongEnough(len) @@ -2037,7 +2049,6 @@ void MyMesh::onContactResponse(const ContactInfo &contact, const uint8_t *data, "onContactResponse(), invalid status response size: len=%u, expected>=%u", (unsigned)len, (unsigned)mesh::COMPANION_MIN_STATUS_RESPONSE_SIZE); - clearPendingReqs(); return; } @@ -2048,11 +2059,9 @@ void MyMesh::onContactResponse(const ContactInfo &contact, const uint8_t *data, i += 6; // pub_key_prefix memcpy(&out_frame[i], &data[4], len - 4); i += (len - 4); - writePendingSerialFrame(out_frame, i); - clearPendingReqs(); - } else if (pending_telemetry && len > 4 && tag == pending_telemetry) { // check for matching response tag - if (len > MAX_FRAME_SIZE - 4) { - clearPendingReqs(); + writePendingSerialFrame(out_frame, i, now); + } else if (matched && kind == mesh::CompanionDelayedReplies::Telemetry) { + if (len <= 4 || len > MAX_FRAME_SIZE - 4) { return; } int i = 0; @@ -2062,11 +2071,9 @@ void MyMesh::onContactResponse(const ContactInfo &contact, const uint8_t *data, i += 6; // pub_key_prefix memcpy(&out_frame[i], &data[4], len - 4); i += (len - 4); - writePendingSerialFrame(out_frame, i); - clearPendingReqs(); - } else if (pending_req && len > 4 && tag == pending_req) { // check for matching response tag - if (len > MAX_FRAME_SIZE - 2) { - clearPendingReqs(); + writePendingSerialFrame(out_frame, i, now); + } else if (matched && kind == mesh::CompanionDelayedReplies::Binary) { + if (len <= 4 || len > MAX_FRAME_SIZE - 2) { return; } int i = 0; @@ -2076,18 +2083,14 @@ void MyMesh::onContactResponse(const ContactInfo &contact, const uint8_t *data, i += 4; memcpy(&out_frame[i], &data[4], len - 4); i += (len - 4); - writePendingSerialFrame(out_frame, i); - clearPendingReqs(); + writePendingSerialFrame(out_frame, i, now); } else { // Forward unsolicited responses, including pushed telemetry, to the app. // Pending requests above retain their captured reply route. // A matching tag with no response body is malformed, not an unsolicited // push, and must not escape through the app's generic response path. - if (len == 4 && ((pending_telemetry && tag == pending_telemetry) - || (pending_req && tag == pending_req))) { - clearPendingReqs(); - return; - } + if (matched || _delayed_replies.blocksResponse(contact.id.pub_key, tag, + now)) return; if (_serial != NULL && _serial->isConnected() && len <= MAX_FRAME_SIZE - 2) { int i = 0; out_frame[i++] = PUSH_CODE_BINARY_RESPONSE; @@ -2100,12 +2103,17 @@ void MyMesh::onContactResponse(const ContactInfo &contact, const uint8_t *data, } bool MyMesh::onContactPathRecv(ContactInfo& contact, uint8_t* in_path, uint8_t in_path_len, uint8_t* out_path, uint8_t out_path_len, uint8_t extra_type, uint8_t* extra, uint8_t extra_len) { - if (extra_type == PAYLOAD_TYPE_RESPONSE && extra_len > 4) { + const uint32_t now = _ms->getMillis(); + servicePendingSerialReply(now); + if (extra_type == PAYLOAD_TYPE_RESPONSE && extra != NULL && extra_len >= 4) { uint32_t tag; memcpy(&tag, extra, 4); - if (pending_discovery != 0 && tag == pending_discovery) { // match an active discovery - if (!mesh::Packet::isValidPathLen(in_path_len) || !mesh::Packet::isValidPathLen(out_path_len)) { + if (_delayed_replies.request.kind == mesh::CompanionDelayedReplies::Discovery + && _delayed_replies.requestMatches(contact.id.pub_key, tag)) { + if (_delayed_replies.request.phase != mesh::CompanionDelayedReplies::AwaitRadio) return false; + if (extra_len <= 4 || !mesh::Packet::isValidPathLen(in_path_len) + || !mesh::Packet::isValidPathLen(out_path_len)) { MESH_DEBUG_PRINTLN("onContactPathRecv, invalid path sizes: %d, %d", in_path_len, out_path_len); } else { int i = 0; @@ -2119,9 +2127,8 @@ bool MyMesh::onContactPathRecv(ContactInfo& contact, uint8_t* in_path, uint8_t i i += mesh::Packet::writePath(&out_frame[i], in_path, in_path_len); // NOTE: telemetry data in 'extra' is discarded at present - writePendingSerialFrame(out_frame, i); + writePendingSerialFrame(out_frame, i, now); } - clearPendingReqs(); return false; // DON'T send reciprocal path! } } @@ -2174,12 +2181,19 @@ void MyMesh::onRawDataRecv(mesh::Packet *packet) { void MyMesh::onTraceRecv(mesh::Packet *packet, uint32_t tag, uint32_t auth_code, uint8_t flags, const uint8_t *path_snrs, const uint8_t *path_hashes, uint8_t path_len) { - const bool binary_trace_match = binary_trace_pending - && tag == binary_trace_tag && auth_code == binary_trace_auth; + const uint32_t now = _ms->getMillis(); + serviceBinaryTraceReply(now); +#if COMPANION_FEATURE_TEXT_TERMINAL + serviceTerminalTrace(now); +#endif + if (_delayed_replies.blocksTrace(tag, auth_code, now)) return; + const bool binary_trace_match = _delayed_replies.hasBinaryTrace() + && tag == _delayed_replies.trace.tag && auth_code == _delayed_replies.trace.auth; + if (binary_trace_match + && _delayed_replies.trace.phase != mesh::CompanionDelayedReplies::AwaitRadio) return; uint8_t path_sz = flags & 0x03; // NEW v1.11+ if (12 + path_len + (path_len >> path_sz) + 1 > MAX_FRAME_SIZE) { MESH_DEBUG_PRINTLN("onTraceRecv(), path_len is too long: %d", (uint32_t)path_len); - if (binary_trace_match) clearBinaryTraceReply(); return; } int i = 0; @@ -2205,14 +2219,14 @@ void MyMesh::onTraceRecv(mesh::Packet *packet, uint32_t tag, uint32_t auth_code, const uint8_t hash_size = _terminal_trace_hash_size; const uint8_t hop_count = hash_size == 0 ? 0 : path_len / hash_size; const uint8_t response_hash_size = 1 << (flags & 0x03); - const unsigned long elapsed = _ms->getMillis() - _terminal_trace_sent_at; - output.printf("\r\nTRACE -> %s (%lu ms)\r\n", - _terminal_trace_target, elapsed); + const unsigned long elapsed = (uint32_t)(now - _terminal_trace_sent_at); if (hash_size == 0 || response_hash_size != hash_size || path_len % hash_size != 0 || hop_count >= MAX_PATH_SIZE) { - output.print(" ERROR: malformed trace response\r\n> "); + return; // A malformed match must not consume a later valid response. } else { + output.printf("\r\nTRACE -> %s (%lu ms)\r\n", + _terminal_trace_target, elapsed); output.print(" "); for (uint8_t hop = 0; hop < hop_count; hop++) { output.print(((float)(int8_t)path_snrs[hop]) / 4.0f, 2); @@ -2224,13 +2238,13 @@ void MyMesh::onTraceRecv(mesh::Packet *packet, uint32_t tag, uint32_t auth_code, output.print(packet->getSNR(), 2); output.print(" dB\r\n> "); } - clearTerminalTrace(); + clearTerminalTrace(now); } #endif if (binary_trace_match) { - _serial->writeFrameToRoute(binary_trace_reply_route, out_frame, i); - clearBinaryTraceReply(); + _delayed_replies.storeBinaryTrace(out_frame, i, now); + serviceBinaryTraceReply(now); } } @@ -2273,6 +2287,7 @@ MyMesh::MyMesh(mesh::Radio &radio, mesh::RNG &rng, mesh::RTCClock &rtc, SimpleMe _terminal_login_expires_at = 0; _terminal_login_target[0] = 0; clearTerminalCommand(); + _terminal_trace_pending = false; clearTerminalTrace(); #endif saved_radio_apply_pending = false; @@ -2287,11 +2302,6 @@ MyMesh::MyMesh(mesh::Radio &radio, mesh::RNG &rng, mesh::RTCClock &rtc, SimpleMe command_radio_repeat = 0; command_radio_apply_deadline = 0; command_radio_reply_route = NULL; - binary_trace_pending = false; - binary_trace_tag = 0; - binary_trace_auth = 0; - binary_trace_deadline = 0; - binary_trace_reply_route = NULL; _scheduled_reboot_at = 0; #if COMPANION_FEATURE_TEMP_RADIO _temp_radio_set_at = 0; @@ -2312,7 +2322,6 @@ MyMesh::MyMesh(mesh::Radio &radio, mesh::RNG &rng, mesh::RTCClock &rtc, SimpleMe offline_queue_capacity = OFFLINE_QUEUE_PSRAM_FALLBACK_SIZE; #endif app_target_ver = 0; - clearPendingReqs(); memset(expected_ack_table, 0, sizeof(expected_ack_table)); next_ack_idx = 0; next_ack_expiry = 0; @@ -5075,22 +5084,52 @@ void MyMesh::startInterface(BaseSerialInterface &serial) { } void MyMesh::clearPendingReqs() { - pending_login = pending_status = pending_telemetry = pending_discovery = - pending_req = 0; - pending_serial_reply_route = NULL; - pending_serial_reply_deadline = 0; + _delayed_replies.retireRequest(_ms->getMillis()); } bool MyMesh::hasPendingReqs() const { - return pending_login != 0 || pending_status != 0 || pending_telemetry != 0 - || pending_discovery != 0 || pending_req != 0; + return _delayed_replies.hasRequest(); +} + +bool MyMesh::beginPendingRequest(mesh::CompanionDelayedReplies::Kind kind, + const ContactInfo& contact, bool terminal) { + servicePendingSerialReply(); + _request_tag_rejected = false; + return _delayed_replies.reserveRequest(kind, contact.id.pub_key, + terminal || _serial == NULL ? NULL : _serial->captureReplyRoute(), + terminal, _ms->getMillis()); +} + +bool MyMesh::allowRequestTag(uint32_t tag) { + // Other internal Base callers need no host correlation reservation. + if (_delayed_replies.request.phase != mesh::CompanionDelayedReplies::Reserved) return true; + _request_tag_rejected = !_delayed_replies.allowRequestTag(tag); + return !_request_tag_rejected; +} + +void MyMesh::armPendingRequest(uint32_t tag, uint32_t timeout, bool flood) { + const uint32_t now = _ms->getMillis(); + _delayed_replies.armRequest(tag, timeout, flood, now); + servicePendingSerialReply(now); +} + +void MyMesh::finishPendingRequest(int result, uint32_t tag, uint32_t timeout) { + if (result == MSG_SEND_FAILED) { + abandonPendingRequest(); + writeErrFrame(_request_tag_rejected ? ERR_CODE_BAD_STATE : ERR_CODE_TABLE_FULL); + } else { + armPendingRequest(tag, timeout, result == MSG_SEND_SENT_FLOOD); + } +} + +void MyMesh::abandonPendingRequest() { + _delayed_replies.abandonRequest(); } bool MyMesh::hasFiniteDelayedReplyForRoute(BaseSerialInterface* route) const { if (route == NULL) return false; - if (pending_serial_reply_route == route - || (command_radio_apply_pending && command_radio_reply_route == route) - || (binary_trace_pending && binary_trace_reply_route == route)) { + if (_delayed_replies.hasReplyForRoute(route, _ms->getMillis()) + || (command_radio_apply_pending && command_radio_reply_route == route)) { return true; } for (int i = 0; i < EXPECTED_ACK_TABLE_SIZE; ++i) { @@ -5103,29 +5142,26 @@ bool MyMesh::hasFiniteDelayedReplyForRoute(BaseSerialInterface* route) const { } void MyMesh::servicePendingSerialReply() { - if (pending_serial_reply_route == NULL) return; - if (!_serial->isReplyRouteAvailable(pending_serial_reply_route) - || pending_serial_reply_deadline == _ms->getMillis() - || millisHasNowPassed(pending_serial_reply_deadline)) { - clearPendingReqs(); - } + servicePendingSerialReply(_ms->getMillis()); +} + +void MyMesh::servicePendingSerialReply(uint32_t now) { +#if COMPANION_FEATURE_TEXT_TERMINAL + serviceTerminalLogin(now); +#endif + _delayed_replies.serviceRequest(_serial, now); } void MyMesh::clearBinaryTraceReply() { - binary_trace_pending = false; - binary_trace_tag = 0; - binary_trace_auth = 0; - binary_trace_deadline = 0; - binary_trace_reply_route = NULL; + _delayed_replies.retireBinaryTrace(_ms->getMillis()); } void MyMesh::serviceBinaryTraceReply() { - if (!binary_trace_pending) return; - if (!_serial->isReplyRouteAvailable(binary_trace_reply_route) - || binary_trace_deadline == _ms->getMillis() - || millisHasNowPassed(binary_trace_deadline)) { - clearBinaryTraceReply(); - } + serviceBinaryTraceReply(_ms->getMillis()); +} + +void MyMesh::serviceBinaryTraceReply(uint32_t now) { + _delayed_replies.serviceBinaryTrace(_serial, now); } void MyMesh::cancelSigningSession() { @@ -5157,9 +5193,9 @@ void MyMesh::cancelSerialOperationsForRoute(BaseSerialInterface* route) { memset(private_key_backup_nonce, 0, sizeof(private_key_backup_nonce)); memset(private_key_backup_sender, 0, sizeof(private_key_backup_sender)); } - if (pending_serial_reply_route == route) clearPendingReqs(); + if (_delayed_replies.request.route == route) clearPendingReqs(); if (command_radio_reply_route == route) cancelPendingRadioParamApply(); - if (binary_trace_reply_route == route) clearBinaryTraceReply(); + if (_delayed_replies.trace.route == route) clearBinaryTraceReply(); if (sign_data_reply_route == route) cancelSigningSession(); for (int i = 0; i < EXPECTED_ACK_TABLE_SIZE; ++i) { if (expected_ack_table[i].reply_route == route) { @@ -5185,6 +5221,7 @@ void MyMesh::handleCmdFrame(size_t len) { return; } + servicePendingSerialReply(); const bool starts_long_lived_request = (cmd_frame[0] == CMD_SEND_LOGIN && len >= 1 + PUB_KEY_SIZE) || (cmd_frame[0] == CMD_SEND_ANON_REQ && len > 1 + PUB_KEY_SIZE) @@ -6010,22 +6047,14 @@ void MyMesh::handleCmdFrame(size_t len) { char *password = (char *)&cmd_frame[1 + PUB_KEY_SIZE]; cmd_frame[len] = 0; // ensure null terminator in password if (recipient) { - uint32_t est_timeout; - int result = sendLogin(*recipient, password, est_timeout); - if (result == MSG_SEND_FAILED) { - writeErrFrame(ERR_CODE_TABLE_FULL); - } else { - clearPendingReqs(); - pending_serial_reply_route = _serial->captureReplyRoute(); - pending_serial_reply_deadline = - futureMillis(est_timeout + est_timeout / 5); - memcpy(&pending_login, recipient->id.pub_key, 4); // match this to onContactResponse() - out_frame[0] = RESP_CODE_SENT; - out_frame[1] = (result == MSG_SEND_SENT_FLOOD) ? 1 : 0; - memcpy(&out_frame[2], &pending_login, 4); - memcpy(&out_frame[6], &est_timeout, 4); - _serial->writeFrame(out_frame, 10); + if (!beginPendingRequest(mesh::CompanionDelayedReplies::Login, *recipient)) { + writeErrFrame(ERR_CODE_BAD_STATE); + return; } + uint32_t tag, est_timeout = 0; + memcpy(&tag, recipient->id.pub_key, 4); + int result = sendLogin(*recipient, password, est_timeout); + finishPendingRequest(result, tag, est_timeout); } else { writeErrFrame(ERR_CODE_NOT_FOUND); // contact not found } @@ -6047,22 +6076,13 @@ void MyMesh::handleCmdFrame(size_t len) { } uint8_t *data = &cmd_frame[1 + PUB_KEY_SIZE]; if (recipient) { - uint32_t tag, est_timeout; - int result = sendAnonReq(*recipient, data, len - (1 + PUB_KEY_SIZE), tag, est_timeout); - if (result == MSG_SEND_FAILED) { - writeErrFrame(ERR_CODE_TABLE_FULL); - } else { - clearPendingReqs(); - pending_serial_reply_route = _serial->captureReplyRoute(); - pending_serial_reply_deadline = - futureMillis(est_timeout + est_timeout / 5); - pending_req = tag; // match this to onContactResponse() - out_frame[0] = RESP_CODE_SENT; - out_frame[1] = (result == MSG_SEND_SENT_FLOOD) ? 1 : 0; - memcpy(&out_frame[2], &tag, 4); - memcpy(&out_frame[6], &est_timeout, 4); - _serial->writeFrame(out_frame, 10); + if (!beginPendingRequest(mesh::CompanionDelayedReplies::Binary, *recipient)) { + writeErrFrame(ERR_CODE_BAD_STATE); + return; } + uint32_t tag = 0, est_timeout = 0; + int result = sendAnonReq(*recipient, data, len - (1 + PUB_KEY_SIZE), tag, est_timeout); + finishPendingRequest(result, tag, est_timeout); } else { writeErrFrame(ERR_CODE_TABLE_FULL); // contacts full } @@ -6070,22 +6090,13 @@ void MyMesh::handleCmdFrame(size_t len) { uint8_t *pub_key = &cmd_frame[1]; ContactInfo *recipient = lookupContactByPubKey(pub_key, PUB_KEY_SIZE); if (recipient) { - uint32_t tag, est_timeout; - int result = sendRequest(*recipient, REQ_TYPE_GET_STATUS, tag, est_timeout); - if (result == MSG_SEND_FAILED) { - writeErrFrame(ERR_CODE_TABLE_FULL); - } else { - clearPendingReqs(); - pending_serial_reply_route = _serial->captureReplyRoute(); - pending_serial_reply_deadline = - futureMillis(est_timeout + est_timeout / 5); - pending_status = tag; // match the reflected tag in onContactResponse() - out_frame[0] = RESP_CODE_SENT; - out_frame[1] = (result == MSG_SEND_SENT_FLOOD) ? 1 : 0; - memcpy(&out_frame[2], &tag, 4); - memcpy(&out_frame[6], &est_timeout, 4); - _serial->writeFrame(out_frame, 10); + if (!beginPendingRequest(mesh::CompanionDelayedReplies::Status, *recipient)) { + writeErrFrame(ERR_CODE_BAD_STATE); + return; } + uint32_t tag = 0, est_timeout = 0; + int result = sendRequest(*recipient, REQ_TYPE_GET_STATUS, tag, est_timeout); + finishPendingRequest(result, tag, est_timeout); } else { writeErrFrame(ERR_CODE_NOT_FOUND); // contact not found } @@ -6093,7 +6104,11 @@ void MyMesh::handleCmdFrame(size_t len) { uint8_t *pub_key = &cmd_frame[2]; ContactInfo *recipient = lookupContactByPubKey(pub_key, PUB_KEY_SIZE); if (recipient) { - uint32_t tag, est_timeout; + if (!beginPendingRequest(mesh::CompanionDelayedReplies::Discovery, *recipient)) { + writeErrFrame(ERR_CODE_BAD_STATE); + return; + } + uint32_t tag = 0, est_timeout = 0; // 'Path Discovery' is just a special case of flood + Telemetry req uint8_t req_data[9]; req_data[0] = REQ_TYPE_GET_TELEMETRY_DATA; @@ -6104,20 +6119,7 @@ void MyMesh::handleCmdFrame(size_t len) { recipient->out_path_len = OUT_PATH_UNKNOWN; int result = sendRequest(*recipient, req_data, sizeof(req_data), tag, est_timeout); recipient->out_path_len = save; - if (result == MSG_SEND_FAILED) { - writeErrFrame(ERR_CODE_TABLE_FULL); - } else { - clearPendingReqs(); - pending_serial_reply_route = _serial->captureReplyRoute(); - pending_serial_reply_deadline = - futureMillis(est_timeout + est_timeout / 5); - pending_discovery = tag; // match this in onContactResponse() - out_frame[0] = RESP_CODE_SENT; - out_frame[1] = (result == MSG_SEND_SENT_FLOOD) ? 1 : 0; - memcpy(&out_frame[2], &tag, 4); - memcpy(&out_frame[6], &est_timeout, 4); - _serial->writeFrame(out_frame, 10); - } + finishPendingRequest(result, tag, est_timeout); } else { writeErrFrame(ERR_CODE_NOT_FOUND); // contact not found } @@ -6125,22 +6127,13 @@ void MyMesh::handleCmdFrame(size_t len) { uint8_t *pub_key = &cmd_frame[4]; ContactInfo *recipient = lookupContactByPubKey(pub_key, PUB_KEY_SIZE); if (recipient) { - uint32_t tag, est_timeout; - int result = sendRequest(*recipient, REQ_TYPE_GET_TELEMETRY_DATA, tag, est_timeout); - if (result == MSG_SEND_FAILED) { - writeErrFrame(ERR_CODE_TABLE_FULL); - } else { - clearPendingReqs(); - pending_serial_reply_route = _serial->captureReplyRoute(); - pending_serial_reply_deadline = - futureMillis(est_timeout + est_timeout / 5); - pending_telemetry = tag; // match this in onContactResponse() - out_frame[0] = RESP_CODE_SENT; - out_frame[1] = (result == MSG_SEND_SENT_FLOOD) ? 1 : 0; - memcpy(&out_frame[2], &tag, 4); - memcpy(&out_frame[6], &est_timeout, 4); - _serial->writeFrame(out_frame, 10); + if (!beginPendingRequest(mesh::CompanionDelayedReplies::Telemetry, *recipient)) { + writeErrFrame(ERR_CODE_BAD_STATE); + return; } + uint32_t tag = 0, est_timeout = 0; + int result = sendRequest(*recipient, REQ_TYPE_GET_TELEMETRY_DATA, tag, est_timeout); + finishPendingRequest(result, tag, est_timeout); } else { writeErrFrame(ERR_CODE_NOT_FOUND); // contact not found } @@ -6169,22 +6162,13 @@ void MyMesh::handleCmdFrame(size_t len) { ContactInfo *recipient = lookupContactByPubKey(pub_key, PUB_KEY_SIZE); if (recipient) { uint8_t *req_data = &cmd_frame[1 + PUB_KEY_SIZE]; - uint32_t tag, est_timeout; - int result = sendRequest(*recipient, req_data, len - (1 + PUB_KEY_SIZE), tag, est_timeout); - if (result == MSG_SEND_FAILED) { - writeErrFrame(ERR_CODE_TABLE_FULL); - } else { - clearPendingReqs(); - pending_serial_reply_route = _serial->captureReplyRoute(); - pending_serial_reply_deadline = - futureMillis(est_timeout + est_timeout / 5); - pending_req = tag; // match this in onContactResponse() - out_frame[0] = RESP_CODE_SENT; - out_frame[1] = (result == MSG_SEND_SENT_FLOOD) ? 1 : 0; - memcpy(&out_frame[2], &tag, 4); - memcpy(&out_frame[6], &est_timeout, 4); - _serial->writeFrame(out_frame, 10); + if (!beginPendingRequest(mesh::CompanionDelayedReplies::Binary, *recipient)) { + writeErrFrame(ERR_CODE_BAD_STATE); + return; } + uint32_t tag = 0, est_timeout = 0; + int result = sendRequest(*recipient, req_data, len - (1 + PUB_KEY_SIZE), tag, est_timeout); + finishPendingRequest(result, tag, est_timeout); } else { writeErrFrame(ERR_CODE_NOT_FOUND); // contact not found } @@ -6292,7 +6276,7 @@ void MyMesh::handleCmdFrame(size_t len) { } } else if (cmd_frame[0] == CMD_SEND_TRACE_PATH && len > 10 && len - 10 <= MAX_PACKET_PAYLOAD - 9) { serviceBinaryTraceReply(); - if (binary_trace_pending) { + if (_delayed_replies.hasBinaryTrace()) { writeErrFrame(ERR_CODE_BAD_STATE); return; } @@ -6305,6 +6289,11 @@ void MyMesh::handleCmdFrame(size_t len) { uint32_t tag, auth; memcpy(&tag, &cmd_frame[1], 4); memcpy(&auth, &cmd_frame[5], 4); + if (!_delayed_replies.reserveBinaryTrace(tag, auth, + _serial->captureReplyRoute(), _ms->getMillis())) { + writeErrFrame(ERR_CODE_BAD_STATE); + return; + } auto pkt = createTrace(tag, auth, flags); if (pkt) { // Compute before handing ownership to sendDirect(), which releases the @@ -6312,21 +6301,14 @@ void MyMesh::handleCmdFrame(size_t len) { uint32_t t = _radio->getEstAirtimeFor(9 + path_len + 2); uint32_t est_timeout = calcDirectTimeoutMillisFor(t, path_len >> path_sz); if (sendDirect(pkt, &cmd_frame[10], path_len)) { - binary_trace_pending = true; - binary_trace_tag = tag; - binary_trace_auth = auth; - binary_trace_deadline = - futureMillis(est_timeout + est_timeout / 5); - binary_trace_reply_route = _serial->captureReplyRoute(); - out_frame[0] = RESP_CODE_SENT; - out_frame[1] = 0; - memcpy(&out_frame[2], &tag, 4); - memcpy(&out_frame[6], &est_timeout, 4); - _serial->writeFrame(out_frame, 10); + _delayed_replies.armBinaryTrace(est_timeout, _ms->getMillis()); + serviceBinaryTraceReply(); } else { + _delayed_replies.abandonBinaryTrace(); writeErrFrame(ERR_CODE_TABLE_FULL); } } else { + _delayed_replies.abandonBinaryTrace(); writeErrFrame(ERR_CODE_TABLE_FULL); } } @@ -8494,10 +8476,14 @@ void MyMesh::printTerminalSendStatus(const char* operation, } void MyMesh::clearTerminalLogin() { - if (_terminal_login_pending && pending_login != 0 - && memcmp(&pending_login, _terminal_login_key, - sizeof(_terminal_login_key)) == 0) { - pending_login = 0; + clearTerminalLogin(_ms->getMillis()); +} + +void MyMesh::clearTerminalLogin(uint32_t now) { + if (_terminal_login_pending && _delayed_replies.request.terminal + && _delayed_replies.request.kind == mesh::CompanionDelayedReplies::Login + && _delayed_replies.requestMatches(_terminal_login_key, 0)) { + _delayed_replies.retireRequest(now); } _terminal_login_pending = false; memset(_terminal_login_key, 0, sizeof(_terminal_login_key)); @@ -8506,10 +8492,12 @@ void MyMesh::clearTerminalLogin() { } void MyMesh::serviceTerminalLogin() { + serviceTerminalLogin(_ms->getMillis()); +} + +void MyMesh::serviceTerminalLogin(uint32_t now) { if (!_terminal_login_pending) return; - const unsigned long now = _ms->getMillis(); - if (_terminal_login_expires_at != now - && !millisHasNowPassed(_terminal_login_expires_at)) { + if ((int32_t)(now - (uint32_t)_terminal_login_expires_at) < 0) { return; } @@ -8517,12 +8505,12 @@ void MyMesh::serviceTerminalLogin() { terminalOutput().printf("\r\n ERROR: login to %s timed out (wrong password or no response).\r\n> ", _terminal_login_target); } - clearTerminalLogin(); + clearTerminalLogin(now); } void MyMesh::sendTerminalLogin(ContactInfo& recipient, const char* password) { - serviceTerminalLogin(); + servicePendingSerialReply(); if (_terminal_login_pending) { terminalOutput().printf(" ERROR: login to %s is still pending\r\n", _terminal_login_target); @@ -8540,20 +8528,27 @@ void MyMesh::sendTerminalLogin(ContactInfo& recipient, return; } + if (!beginPendingRequest(mesh::CompanionDelayedReplies::Login, recipient, true)) { + terminalOutput().print(" ERROR: login is quarantined or reply history is full\r\n"); + return; + } uint32_t est_timeout = 0; const int result = sendLogin(recipient, password, est_timeout); if (result == MSG_SEND_FAILED) { + abandonPendingRequest(); terminalOutput().print(" ERROR: unable to send login\r\n"); return; } - clearPendingReqs(); - memcpy(&pending_login, recipient.id.pub_key, sizeof(pending_login)); + uint32_t tag; + memcpy(&tag, recipient.id.pub_key, 4); + _delayed_replies.armRequest(tag, est_timeout, + result == MSG_SEND_SENT_FLOOD, _ms->getMillis()); _terminal_login_pending = true; memcpy(_terminal_login_key, recipient.id.pub_key, sizeof(_terminal_login_key)); const uint32_t timeout = est_timeout + est_timeout / 5; - _terminal_login_expires_at = futureMillis(timeout); + _terminal_login_expires_at = _delayed_replies.request.radio_deadline; StrHelper::strzcpy(_terminal_login_target, recipient.name, sizeof(_terminal_login_target)); printTerminalSendStatus("Login", recipient, result, timeout); @@ -8639,6 +8634,15 @@ void MyMesh::sendTerminalCommand(ContactInfo& recipient, } void MyMesh::clearTerminalTrace() { + clearTerminalTrace(_ms->getMillis()); +} + +void MyMesh::clearTerminalTrace(uint32_t now) { + if (_terminal_trace_pending) { + _delayed_replies.retireTrace(_terminal_trace_history, now, + _terminal_trace_expires_at); + } + _terminal_trace_history = mesh::CompanionDelayedReplies::NO_HISTORY; _terminal_trace_pending = false; _terminal_trace_hash_size = 0; _terminal_trace_tag = 0; @@ -8649,10 +8653,12 @@ void MyMesh::clearTerminalTrace() { } void MyMesh::serviceTerminalTrace() { + serviceTerminalTrace(_ms->getMillis()); +} + +void MyMesh::serviceTerminalTrace(uint32_t now) { if (!_terminal_trace_pending) return; - const unsigned long now = _ms->getMillis(); - if (_terminal_trace_expires_at != now - && !millisHasNowPassed(_terminal_trace_expires_at)) { + if ((int32_t)(now - (uint32_t)_terminal_trace_expires_at) < 0) { return; } @@ -8660,7 +8666,7 @@ void MyMesh::serviceTerminalTrace() { terminalOutput().printf("\r\n ERROR: trace to %s timed out.\r\n> ", _terminal_trace_target); } - clearTerminalTrace(); + clearTerminalTrace(now); } void MyMesh::sendTerminalTraceRoute(const uint8_t* route, uint8_t hash_size, @@ -8697,7 +8703,9 @@ void MyMesh::sendTerminalTraceRoute(const uint8_t* route, uint8_t hash_size, calcDirectTimeoutMillisFor(airtime, hop_count); uint32_t trace_timeout = 0; if (!mesh::calculateTerminalTraceTimeoutMillis(base_timeout, - trace_timeout)) { + trace_timeout) + || trace_timeout > 0x7FFFFFFFU + - mesh::CompanionDelayedReplies::DELIVERY_GRACE_MS) { terminalOutput().print(" ERROR: trace timeout is out of range\r\n"); return; } @@ -8706,18 +8714,26 @@ void MyMesh::sendTerminalTraceRoute(const uint8_t* route, uint8_t hash_size, uint32_t auth = 0; getRNG()->random((uint8_t*)&tag, sizeof(tag)); getRNG()->random((uint8_t*)&auth, sizeof(auth)); + const uint8_t history = _delayed_replies.reserveTrace(tag, auth, _ms->getMillis()); + if (history == mesh::CompanionDelayedReplies::NO_HISTORY) { + terminalOutput().print(" ERROR: trace tag is quarantined or reply history is full\r\n"); + return; + } mesh::Packet* packet = createTrace(tag, auth, flags); if (packet == NULL) { + _delayed_replies.abandonTrace(history); terminalOutput().print(" ERROR: unable to allocate trace packet\r\n"); return; } if (!sendDirect(packet, route, static_cast(route_byte_len))) { + _delayed_replies.abandonTrace(history); terminalOutput().print(" ERROR: unable to send trace\r\n"); return; } _terminal_trace_pending = true; + _terminal_trace_history = history; _terminal_trace_hash_size = hash_size; _terminal_trace_tag = tag; _terminal_trace_auth = auth; @@ -10324,8 +10340,11 @@ bool MyMesh::canRecoverUsbLogging() const { return dirty_contacts_expiry == 0 && !_store->hasPendingContactWrites() && !hasOutbound() && !isAnyTempRadioActive() && !hasPendingOtaApply() && !command_radio_apply_pending && !saved_radio_apply_pending - && _scheduled_reboot_at == 0 && pending_serial_reply_route == NULL - && !binary_trace_pending && sign_data == NULL + && _scheduled_reboot_at == 0 && !hasPendingReqs() + && !_delayed_replies.hasBinaryTrace() && sign_data == NULL +#if COMPANION_FEATURE_TEXT_TERMINAL + && !_terminal_trace_pending +#endif && !(_serial && _serial->hasPendingIO()) && !(_iter_started && _serial && _serial->isConnected()); } @@ -10355,8 +10374,11 @@ bool MyMesh::hasPendingWork() const { || hasPendingOtaApply() || command_radio_apply_pending || _scheduled_reboot_at != 0 - || pending_serial_reply_route != NULL - || binary_trace_pending + || hasPendingReqs() + || _delayed_replies.hasBinaryTrace() +#if COMPANION_FEATURE_TEXT_TERMINAL + || _terminal_trace_pending +#endif || sign_data != NULL || hasQueuedWorkDue() || hasRetryWorkDue() || (saved_radio_apply_pending diff --git a/examples/companion_radio/MyMesh.h b/examples/companion_radio/MyMesh.h index 2f6c689be..4b3c7a405 100644 --- a/examples/companion_radio/MyMesh.h +++ b/examples/companion_radio/MyMesh.h @@ -42,6 +42,7 @@ #include #include #include +#include #include #include #include @@ -414,6 +415,7 @@ protected: uint32_t calcFloodTimeoutMillisFor(uint32_t pkt_airtime_millis) const override; uint32_t calcDirectTimeoutMillisFor(uint32_t pkt_airtime_millis, uint8_t path_len) const override; void onSendTimeout() override; + bool allowRequestTag(uint32_t tag) override; // DataStoreHost methods bool onContactLoaded(const ContactInfo& contact) override { return addContact(contact); } @@ -477,7 +479,7 @@ private: void writeOKFrame(BaseSerialInterface* route = nullptr); void writeErrFrame(uint8_t err_code, BaseSerialInterface* route = nullptr); - size_t writePendingSerialFrame(const uint8_t frame[], size_t len); + size_t writePendingSerialFrame(const uint8_t frame[], size_t len, uint32_t now); void writeDisabledFrame(); bool writeContactRespFrame(uint8_t code, const ContactInfo &contact); void stopContactsIterator(); @@ -544,13 +546,17 @@ private: const ContactInfo& recipient, int result, uint32_t timeout_millis); void clearTerminalLogin(); + void clearTerminalLogin(uint32_t now); void serviceTerminalLogin(); + void serviceTerminalLogin(uint32_t now); void sendTerminalLogin(ContactInfo& recipient, const char* password); void clearTerminalCommand(); void serviceTerminalCommand(); void sendTerminalCommand(ContactInfo& recipient, const char* command); void clearTerminalTrace(); + void clearTerminalTrace(uint32_t now); void serviceTerminalTrace(); + void serviceTerminalTrace(uint32_t now); void sendTerminalTraceRoute(const uint8_t* route, uint8_t hash_size, uint8_t hop_count, const char* target); void sendTerminalTrace(ContactInfo& recipient); @@ -568,8 +574,15 @@ private: void cancelPendingRadioParamApply(); void servicePendingRadioParamApply(); void servicePendingSerialReply(); + void servicePendingSerialReply(uint32_t now); + bool beginPendingRequest(mesh::CompanionDelayedReplies::Kind kind, + const ContactInfo& contact, bool terminal = false); + void armPendingRequest(uint32_t tag, uint32_t timeout, bool flood); + void finishPendingRequest(int result, uint32_t tag, uint32_t timeout); + void abandonPendingRequest(); void clearBinaryTraceReply(); void serviceBinaryTraceReply(); + void serviceBinaryTraceReply(uint32_t now); void cancelSigningSession(); void serviceSigningSession(); #if COMPANION_FEATURE_TEMP_RADIO @@ -626,12 +639,8 @@ private: bool _wc_mqtt_dirty; bool _wc_batch_active = false; #endif - uint32_t pending_login; - uint32_t pending_status; - uint32_t pending_telemetry, pending_discovery; // pending _TELEMETRY_REQ - uint32_t pending_req; // pending _BINARY_REQ - BaseSerialInterface* pending_serial_reply_route; - unsigned long pending_serial_reply_deadline; + mesh::CompanionDelayedReplies _delayed_replies; + bool _request_tag_rejected = false; BaseSerialInterface* private_key_backup_route = nullptr; unsigned long private_key_backup_deadline = 0; char private_key_backup_nonce[17] = {}; @@ -661,12 +670,13 @@ private: uint8_t _terminal_recipient_key[PUB_KEY_SIZE]; uint8_t _terminal_tmp_buf[MAX_TRANS_UNIT]; bool _terminal_login_pending; - uint8_t _terminal_login_key[4]; + uint8_t _terminal_login_key[PUB_KEY_SIZE]; unsigned long _terminal_login_expires_at; char _terminal_login_target[32]; mesh::TerminalCommandTracker _terminal_command; char _terminal_command_target[32]; bool _terminal_trace_pending; + uint8_t _terminal_trace_history = mesh::CompanionDelayedReplies::NO_HISTORY; uint8_t _terminal_trace_hash_size; uint32_t _terminal_trace_tag; uint32_t _terminal_trace_auth; @@ -686,11 +696,6 @@ private: uint8_t command_radio_repeat; unsigned long command_radio_apply_deadline; BaseSerialInterface* command_radio_reply_route; - bool binary_trace_pending; - uint32_t binary_trace_tag; - uint32_t binary_trace_auth; - unsigned long binary_trace_deadline; - BaseSerialInterface* binary_trace_reply_route; // Deferred so USB/TCP terminals can transmit the acknowledgement before // the transport disappears. Also used by USB interface changes. unsigned long _scheduled_reboot_at; diff --git a/src/helpers/BaseChatMesh.cpp b/src/helpers/BaseChatMesh.cpp index 813d3287e..9af5e346e 100644 --- a/src/helpers/BaseChatMesh.cpp +++ b/src/helpers/BaseChatMesh.cpp @@ -819,6 +819,7 @@ int BaseChatMesh::sendAnonReq(const ContactInfo& recipient, const uint8_t* data, { uint8_t temp[MAX_PACKET_PAYLOAD]; tag = getRTCClock()->getCurrentTimeUnique(); + if (!allowRequestTag(tag)) return MSG_SEND_FAILED; memcpy(temp, &tag, 4); // tag to match later (also extra blob to help make packet_hash unique) memcpy(&temp[4], data, len); @@ -846,6 +847,7 @@ int BaseChatMesh::sendRequest(const ContactInfo& recipient, const uint8_t* req_ { uint8_t temp[MAX_PACKET_PAYLOAD]; tag = getRTCClock()->getCurrentTimeUnique(); + if (!allowRequestTag(tag)) return MSG_SEND_FAILED; memcpy(temp, &tag, 4); // mostly an extra blob to help make packet_hash unique memcpy(&temp[4], req_data, data_len); @@ -871,6 +873,7 @@ int BaseChatMesh::sendRequest(const ContactInfo& recipient, uint8_t req_type, u { uint8_t temp[13]; tag = getRTCClock()->getCurrentTimeUnique(); + if (!allowRequestTag(tag)) return MSG_SEND_FAILED; memcpy(temp, &tag, 4); // mostly an extra blob to help make packet_hash unique temp[4] = req_type; memset(&temp[5], 0, 4); // reserved (possibly for 'since' param) diff --git a/src/helpers/BaseChatMesh.h b/src/helpers/BaseChatMesh.h index dbeabde90..2c61cc047 100644 --- a/src/helpers/BaseChatMesh.h +++ b/src/helpers/BaseChatMesh.h @@ -180,6 +180,8 @@ protected: const uint8_t* data, size_t data_len) {} virtual uint8_t onContactRequest(const ContactInfo& contact, uint32_t sender_timestamp, const uint8_t* data, uint8_t len, uint8_t* reply) = 0; virtual void onContactResponse(const ContactInfo& contact, const uint8_t* data, uint8_t len) = 0; + // Reject a recently retired reflected tag before allocating/transmitting. + virtual bool allowRequestTag(uint32_t tag) { return true; } virtual void handleReturnPathRetry(const ContactInfo& contact, const uint8_t* path, uint8_t path_len); virtual bool sendFloodScoped(const ContactInfo& recipient, mesh::Packet* pkt, uint32_t delay_millis=0); diff --git a/src/helpers/CompanionDelayedReplies.cpp b/src/helpers/CompanionDelayedReplies.cpp new file mode 100644 index 000000000..fc4b7f0ea --- /dev/null +++ b/src/helpers/CompanionDelayedReplies.cpp @@ -0,0 +1,303 @@ +#include "CompanionDelayedReplies.h" +#include + +namespace mesh { + +// Keep complete member initialization shared, not expanded into a large owner. +#if defined(__GNUC__) && !defined(__clang__) +__attribute__((noinline, noclone)) +#elif defined(__clang__) +__attribute__((noinline)) +#endif +CompanionDelayedReplies::CompanionDelayedReplies() noexcept {} + +bool CompanionDelayedReplies::due(uint32_t now, uint32_t deadline) { + return (int32_t)(now - deadline) >= 0; +} + +uint32_t CompanionDelayedReplies::retirementDeadline( + uint32_t now, uint32_t radio_deadline, uint32_t floor) { + uint32_t remaining = due(now, radio_deadline) ? 0 : radio_deadline - now; + if (remaining + DELIVERY_GRACE_MS > floor) floor = remaining + DELIVERY_GRACE_MS; + return now + floor; +} + +void CompanionDelayedReplies::reset(Reply& reply) { + // The frame bytes need not be erased: Empty is the sole ownership authority. + reply.route = nullptr; + reply.kind = None; + reply.phase = Empty; + reply.history = NO_HISTORY; + reply.frame_len = 0; + reply.sent_pending = false; +} + +void CompanionDelayedReplies::arm(Reply& reply, uint32_t timeout, + bool flood, uint32_t now) { + reply.timeout = timeout; + // Wrap-safe deadlines must stay in the signed half of the millis range. + const uint32_t maximum = 0x7FFFFFFFU - DELIVERY_GRACE_MS; + uint32_t budget = timeout > maximum / 6 * 5 ? maximum : timeout + timeout / 5; + reply.radio_deadline = now + budget; + reply.flood = flood; + reply.sent_pending = !reply.terminal; + reply.sent_deadline = now + DELIVERY_GRACE_MS; + reply.frame_len = 0; + reply.phase = AwaitRadio; +} + +bool CompanionDelayedReplies::store(Reply& reply, const uint8_t* frame, + size_t len, uint32_t now) { + if (reply.phase != AwaitRadio || frame == nullptr || len == 0 + || len > sizeof(reply.frame) || due(now, reply.radio_deadline) + || (reply.sent_pending && due(now, reply.sent_deadline))) return false; + memcpy(reply.frame, frame, len); + reply.frame_len = len; + reply.delivery_deadline = now + DELIVERY_GRACE_MS; + if (reply.sent_pending + && (int32_t)(reply.sent_deadline - reply.delivery_deadline) < 0) { + reply.delivery_deadline = reply.sent_deadline; + } + reply.phase = AwaitAdmission; + return true; +} + +bool CompanionDelayedReplies::service(Reply& reply, + BaseSerialInterface* serial, uint32_t now) { + if (reply.phase == Empty || reply.phase == Reserved) return false; + if (reply.terminal) return due(now, reply.radio_deadline); + if (serial == nullptr || !serial->isReplyRouteAvailable(reply.route) + || due(now, reply.phase == AwaitAdmission + ? reply.delivery_deadline : reply.radio_deadline) + || (reply.sent_pending && due(now, reply.sent_deadline))) return true; + if (reply.sent_pending) { + uint8_t sent[10]; + sent[0] = 6; // RESP_CODE_SENT, unchanged Companion wire format. + sent[1] = reply.flood ? 1 : 0; + memcpy(sent + 2, &reply.tag, 4); + memcpy(sent + 6, &reply.timeout, 4); + if (serial->writeFrameToRoute(reply.route, sent, sizeof(sent)) != sizeof(sent)) { + return false; + } + reply.sent_pending = false; + } + return reply.phase == AwaitAdmission + && serial->writeFrameToRoute(reply.route, reply.frame, reply.frame_len) + == reply.frame_len; +} + +void CompanionDelayedReplies::prune(uint32_t now) { + for (unsigned i = 0; i < HISTORY_SIZE; ++i) { + if (requests[i].phase == AwaitAdmission && due(now, requests[i].until)) { + requests[i].phase = Empty; + } + if (traces[i].phase == AwaitAdmission && due(now, traces[i].until)) { + traces[i].phase = Empty; + } + } +} + +bool CompanionDelayedReplies::reserveRequest(Kind kind, const uint8_t peer[32], + BaseSerialInterface* route, bool terminal, uint32_t now) { + prune(now); + if (hasRequest() || kind == None || kind == Trace || peer == nullptr + || (!terminal && route == nullptr)) return false; + uint8_t free = NO_HISTORY; + for (unsigned i = 0; i < HISTORY_SIZE; ++i) { + const RequestGuard& held = requests[i]; + if (held.phase == Empty) { + free = i; + } else if (kind == Login && held.login && memcmp(held.peer, peer, 32) == 0) { + return false; + } + } + if (free == NO_HISTORY) return false; + RequestGuard& held = requests[free]; + memcpy(held.peer, peer, 32); + held.tag = 0; + held.login = kind == Login; + held.phase = Reserved; + request.route = route; + request.kind = kind; + request.history = free; + request.terminal = terminal; + request.radio_deadline = now; + request.phase = Reserved; + return true; +} + +bool CompanionDelayedReplies::allowRequestTag(uint32_t tag) { + if (request.phase != Reserved || request.history >= HISTORY_SIZE) return false; + RequestGuard& held = requests[request.history]; + for (unsigned i = 0; i < HISTORY_SIZE; ++i) { + if (i != request.history && requests[i].phase != Empty + && requests[i].tag == tag && memcmp(requests[i].peer, held.peer, 32) == 0) { + return false; + } + } + held.tag = tag; + return true; +} + +void CompanionDelayedReplies::rememberLoginTag(uint32_t tag) { + if (request.kind == Login && request.history < HISTORY_SIZE) { + requests[request.history].tag = tag; + } +} + +void CompanionDelayedReplies::armRequest(uint32_t tag, uint32_t timeout, + bool flood, uint32_t now) { + request.tag = tag; + if (request.kind != Login && request.history < HISTORY_SIZE) { + requests[request.history].tag = tag; + } + arm(request, timeout, flood, now); +} + +void CompanionDelayedReplies::abandonRequest() { + if (request.history < HISTORY_SIZE) requests[request.history].phase = Empty; + reset(request); +} + +void CompanionDelayedReplies::retireRequest(uint32_t now) { + if (request.phase == Reserved) { + abandonRequest(); + return; + } + if (request.history < HISTORY_SIZE && requests[request.history].phase == Reserved) { + RequestGuard& held = requests[request.history]; + held.until = retirementDeadline(now, request.radio_deadline, + held.login ? LOGIN_QUARANTINE_MS : DELIVERY_GRACE_MS); + held.phase = AwaitAdmission; + } + reset(request); +} + +const uint8_t* CompanionDelayedReplies::requestPeer() const { + return request.history < HISTORY_SIZE ? requests[request.history].peer : nullptr; +} + +bool CompanionDelayedReplies::requestMatches(const uint8_t peer[32], uint32_t tag) const { + const uint8_t* expected = requestPeer(); + return request.phase >= AwaitRadio && expected != nullptr + && memcmp(expected, peer, 32) == 0 + && (request.kind == Login || request.tag == tag); +} + +bool CompanionDelayedReplies::blocksResponse(const uint8_t peer[32], + uint32_t tag, uint32_t now) const { + // A tag-only match from another identity must not become a generic host reply. + if (request.phase >= AwaitRadio && request.kind != Login && request.tag == tag + && (!requestMatches(peer, tag) || request.phase == AwaitAdmission)) return true; + for (unsigned i = 0; i < HISTORY_SIZE; ++i) { + const RequestGuard& held = requests[i]; + if (held.phase == Empty || memcmp(held.peer, peer, 32) != 0) continue; + if (held.phase == AwaitAdmission && due(now, held.until)) continue; + if (i == request.history && request.phase == AwaitRadio) continue; + if (held.login || held.tag == tag) return true; + } + return false; +} + +bool CompanionDelayedReplies::storeRequest(const uint8_t* frame, size_t len, uint32_t now) { + return store(request, frame, len, now); +} + +void CompanionDelayedReplies::serviceRequest(BaseSerialInterface* serial, uint32_t now) { + prune(now); + if (service(request, serial, now)) retireRequest(now); +} + +uint8_t CompanionDelayedReplies::reserveTrace(uint32_t tag, uint32_t auth, uint32_t now) { + prune(now); + uint8_t free = NO_HISTORY; + for (unsigned i = 0; i < HISTORY_SIZE; ++i) { + const TraceGuard& held = traces[i]; + if (held.phase == Empty) { + free = i; + } else if (held.tag == tag && held.auth == auth) { + return NO_HISTORY; + } + } + if (free != NO_HISTORY) { + traces[free].tag = tag; + traces[free].auth = auth; + traces[free].phase = Reserved; + } + return free; +} + +void CompanionDelayedReplies::abandonTrace(uint8_t history) { + if (history < HISTORY_SIZE) traces[history].phase = Empty; +} + +void CompanionDelayedReplies::retireTrace(uint8_t history, uint32_t now, + uint32_t radio_deadline) { + if (history < HISTORY_SIZE && traces[history].phase == Reserved) { + traces[history].until = retirementDeadline(now, radio_deadline, DELIVERY_GRACE_MS); + traces[history].phase = AwaitAdmission; + } +} + +bool CompanionDelayedReplies::reserveBinaryTrace(uint32_t tag, uint32_t auth, + BaseSerialInterface* route, uint32_t now) { + if (hasBinaryTrace() || route == nullptr) return false; + uint8_t history = reserveTrace(tag, auth, now); + if (history == NO_HISTORY) return false; + trace.route = route; + trace.tag = tag; + trace.auth = auth; + trace.kind = Trace; + trace.history = history; + trace.terminal = false; + trace.phase = Reserved; + return true; +} + +void CompanionDelayedReplies::armBinaryTrace(uint32_t timeout, uint32_t now) { + arm(trace, timeout, false, now); +} + +void CompanionDelayedReplies::abandonBinaryTrace() { + abandonTrace(trace.history); + reset(trace); +} + +void CompanionDelayedReplies::retireBinaryTrace(uint32_t now) { + if (trace.phase == Reserved) { + abandonBinaryTrace(); + return; + } + retireTrace(trace.history, now, trace.radio_deadline); + reset(trace); +} + +bool CompanionDelayedReplies::blocksTrace(uint32_t tag, uint32_t auth, uint32_t now) const { + for (unsigned i = 0; i < HISTORY_SIZE; ++i) { + if (traces[i].phase == AwaitAdmission && !due(now, traces[i].until) + && traces[i].tag == tag && traces[i].auth == auth) return true; + } + return false; +} + +bool CompanionDelayedReplies::storeBinaryTrace(const uint8_t* frame, size_t len, uint32_t now) { + return store(trace, frame, len, now); +} + +void CompanionDelayedReplies::serviceBinaryTrace(BaseSerialInterface* serial, uint32_t now) { + prune(now); + if (service(trace, serial, now)) retireBinaryTrace(now); +} + +bool CompanionDelayedReplies::hasReplyForRoute(BaseSerialInterface* route, uint32_t now) const { + const Reply* slots[] = { &request, &trace }; + for (const Reply* reply : slots) { + if (reply->phase >= AwaitRadio && reply->route == route + && !due(now, reply->phase == AwaitAdmission + ? reply->delivery_deadline : reply->radio_deadline) + && !(reply->sent_pending && due(now, reply->sent_deadline))) return true; + } + return false; +} + +} // namespace mesh diff --git a/src/helpers/CompanionDelayedReplies.h b/src/helpers/CompanionDelayedReplies.h new file mode 100644 index 000000000..e7f38c78b --- /dev/null +++ b/src/helpers/CompanionDelayedReplies.h @@ -0,0 +1,93 @@ +#pragma once + +#include "BaseSerialInterface.h" +#include + +namespace mesh { + +// Fixed producer storage, independent of the transport's bounded frame queue. +// History is volatile and finite: it does not authenticate a host process or +// disambiguate arbitrarily late login replies, which have no reflected tag. +class CompanionDelayedReplies { +public: + CompanionDelayedReplies() noexcept; + enum Kind : uint8_t { None, Login, Status, Telemetry, Discovery, Binary, Trace }; + enum Phase : uint8_t { Empty, Reserved, AwaitRadio, AwaitAdmission }; + static constexpr uint8_t HISTORY_SIZE = 8; + static constexpr uint8_t NO_HISTORY = 255; + static constexpr uint32_t DELIVERY_GRACE_MS = 10000; + static constexpr uint32_t LOGIN_QUARANTINE_MS = 30000; + + struct Reply { + BaseSerialInterface* route = nullptr; + uint32_t tag = 0, auth = 0; + uint32_t timeout = 0, radio_deadline = 0; + uint32_t sent_deadline = 0, delivery_deadline = 0; + uint16_t frame_len = 0; + Kind kind = None; + Phase phase = Empty; + uint8_t history = NO_HISTORY; + bool terminal = false, sent_pending = false, flood = false; + uint8_t frame[MAX_FRAME_SIZE] = {}; + }; + + Reply request, trace; + +private: + struct RequestGuard { + uint8_t peer[32] = {}; + uint32_t tag = 0, until = 0; + Phase phase = Empty; + bool login = false; + }; + struct TraceGuard { + uint32_t tag = 0, auth = 0, until = 0; + Phase phase = Empty; + }; + RequestGuard requests[HISTORY_SIZE] = {}; + TraceGuard traces[HISTORY_SIZE] = {}; + + static bool due(uint32_t now, uint32_t deadline); + static uint32_t retirementDeadline(uint32_t now, uint32_t radio_deadline, + uint32_t floor); + static void reset(Reply& reply); + static void arm(Reply& reply, uint32_t timeout, bool flood, uint32_t now); + static bool store(Reply& reply, const uint8_t* frame, size_t len, uint32_t now); + static bool service(Reply& reply, BaseSerialInterface* serial, uint32_t now); + void prune(uint32_t now); + +public: + + bool reserveRequest(Kind kind, const uint8_t peer[32], + BaseSerialInterface* route, bool terminal, uint32_t now); + bool allowRequestTag(uint32_t tag); + void rememberLoginTag(uint32_t tag); + void armRequest(uint32_t tag, uint32_t timeout, bool flood, uint32_t now); + void abandonRequest(); + void retireRequest(uint32_t now); + const uint8_t* requestPeer() const; + bool requestMatches(const uint8_t peer[32], uint32_t tag) const; + bool blocksResponse(const uint8_t peer[32], uint32_t tag, uint32_t now) const; + bool storeRequest(const uint8_t* frame, size_t len, uint32_t now); + void serviceRequest(BaseSerialInterface* serial, uint32_t now); + + // Terminal TRACE reserves the same tuple history without taking the Binary + // slot. A live reservation is never evicted to make room for another host. + uint8_t reserveTrace(uint32_t tag, uint32_t auth, uint32_t now); + void abandonTrace(uint8_t history); + void retireTrace(uint8_t history, uint32_t now, uint32_t radio_deadline); + bool reserveBinaryTrace(uint32_t tag, uint32_t auth, + BaseSerialInterface* route, uint32_t now); + void armBinaryTrace(uint32_t timeout, uint32_t now); + void abandonBinaryTrace(); + void retireBinaryTrace(uint32_t now); + bool blocksTrace(uint32_t tag, uint32_t auth, uint32_t now) const; + bool storeBinaryTrace(const uint8_t* frame, size_t len, uint32_t now); + void serviceBinaryTrace(BaseSerialInterface* serial, uint32_t now); + + bool hasRequest() const { return request.phase != Empty; } + bool hasBinaryTrace() const { return trace.phase != Empty; } + bool hasReplyForRoute(BaseSerialInterface* route, uint32_t now) const; +}; + +} // namespace mesh diff --git a/test/fixtures/companion_ack_delivery/test.cpp b/test/fixtures/companion_ack_delivery/test.cpp index 928f0e207..b0251e026 100644 --- a/test/fixtures/companion_ack_delivery/test.cpp +++ b/test/fixtures/companion_ack_delivery/test.cpp @@ -1,5 +1,6 @@ #include #include +#include #include #include #include @@ -51,15 +52,14 @@ public: const uint8_t* last_connection_ack_input = nullptr; bool retry_active[256] = {}; Terminal terminal; - BaseSerialInterface* pending_serial_reply_route = nullptr; + mesh::CompanionDelayedReplies _delayed_replies; BaseSerialInterface* command_radio_reply_route = nullptr; - BaseSerialInterface* binary_trace_reply_route = nullptr; BaseSerialInterface* sign_data_reply_route = nullptr; BaseSerialInterface* private_key_backup_route = nullptr; unsigned long private_key_backup_deadline = 0; char private_key_backup_nonce[17] = {}; uint8_t private_key_backup_sender[6] = {}; - bool command_radio_apply_pending = false, binary_trace_pending = false; + bool command_radio_apply_pending = false; // Radio dispatch, peer-store, and terminal I/O are hardware boundaries. All // ACK state decisions and transport admission execute production code. bool cancelActiveRetries(const uint8_t* key) { @@ -81,9 +81,9 @@ public: } bool hasTerminalOutput() const { return true; } Terminal& terminalOutput() { return terminal; } - void clearPendingReqs() { pending_serial_reply_route = nullptr; } + void clearPendingReqs() { _delayed_replies.retireRequest(millis()); } void cancelPendingRadioParamApply() { command_radio_reply_route = nullptr; } - void clearBinaryTraceReply() { binary_trace_reply_route = nullptr; } + void clearBinaryTraceReply() { _delayed_replies.retireBinaryTrace(millis()); } void cancelSigningSession() { sign_data_reply_route = nullptr; } bool millisHasNowPassed(unsigned long) const; unsigned long futureMillis(int) const; diff --git a/test/fixtures/companion_delayed_reply_delivery/test.cpp b/test/fixtures/companion_delayed_reply_delivery/test.cpp new file mode 100644 index 000000000..65f80f0db --- /dev/null +++ b/test/fixtures/companion_delayed_reply_delivery/test.cpp @@ -0,0 +1,760 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#define MESH_DEBUG_PRINTLN(...) ((void)0) +#include "production_constants.inc" +#include "production_stream.inc" +std::deque WiFiServer::incoming; + +using Tracker = mesh::CompanionDelayedReplies; +static_assert(std::is_nothrow_default_constructible::value, + "The fixed reply tracker constructor must not add exception cleanup"); +using Kind = Tracker::Kind; +using Bytes = std::vector; +struct Identity { uint8_t pub_key[32] = {1, 2, 3, 4, 5, 6}; }; +struct ContactInfo { + Identity id; + uint8_t type = ADV_TYPE_ROOM, out_path_len = 1; + uint32_t sync_since = 0, lastmod = 0; + char name[32] = "synthetic peer"; + uint8_t path[64] = {0x12}; + const uint8_t* getPath() const { return path; } + bool copyPathTo(uint8_t* target) const { memcpy(target,path,sizeof(path)); return true; } + const uint8_t* getSharedSecret(const Identity&) const { + static const uint8_t synthetic[32] = {}; + return synthetic; + } + bool setPath(const uint8_t*, uint8_t) { return true; } +}; +struct RtcBoundary { + uint32_t wall = 1000, last_unique = 0; + uint32_t getCurrentTime() const { return wall; } +#include "production_clock.inc" +}; +static uint32_t clock_step = 0; +struct Clock { uint32_t getMillis() const { uint32_t now=millis();g_mock_millis+=clock_step;return now; } }; +struct RngBoundary { + unsigned calls = 0; + void random(uint8_t* bytes, size_t len) { + const uint32_t value = (calls++ & 1) ? 23 : 17; + assert(len <= sizeof(value)); + memcpy(bytes, &value, len); + } +}; +struct RadioBoundary { uint32_t getEstAirtimeFor(size_t) const { return 100; } }; +namespace mesh { +struct Packet { + uint8_t payload[256] = {}, payload_len = 0, path_len = 0; + float getSNR() const { return 0; } + static bool isValidPathLen(uint8_t); + static size_t writePath(uint8_t*, const uint8_t*, uint8_t); +}; +struct Utils { + static void printHex(Stream& stream, const uint8_t*, size_t) { stream.print("00"); } +}; +} + +struct BaseChatMesh { + Clock clock; + Clock* _ms = &clock; + RtcBoundary rtc; + RngBoundary rng; + RadioBoundary radio; + RadioBoundary* _radio = &radio; + Identity self_id; + mesh::Packet packet; + unsigned allocations = 0, transmissions = 0, connection_starts = 0; + uint32_t txt_send_timeout = 0; + uint32_t direct_timeout = 5000; + bool radio_accepts = true; + RtcBoundary* getRTCClock() { return &rtc; } + RngBoundary* getRNG() { return &rng; } + virtual bool allowRequestTag(uint32_t) { return true; } + virtual void onContactResponse(const ContactInfo&, const uint8_t*, uint8_t) = 0; + bool canMutateContacts() const { return false; } + void onContactPathUpdated(ContactInfo&) {} + bool processAck(const uint8_t*, ContactInfo*&) { return false; } + bool onContactPathRecv(ContactInfo&, uint8_t*, uint8_t, uint8_t*, uint8_t, + uint8_t, uint8_t*, uint8_t); + mesh::Packet* allocate() { ++allocations; return &packet; } + mesh::Packet* createDatagram(uint8_t, const Identity&, const uint8_t*, const uint8_t*, size_t) { + return allocate(); + } + mesh::Packet* createAnonDatagram(uint8_t, const Identity&, const Identity&, + const uint8_t*, const uint8_t*, size_t) { return allocate(); } + mesh::Packet* createTrace(uint32_t, uint32_t, uint8_t) { return allocate(); } + bool sendDirect(mesh::Packet*, const uint8_t*, uint8_t) { ++transmissions; return radio_accepts; } + bool sendFloodScoped(const ContactInfo&, mesh::Packet*) { ++transmissions; return radio_accepts; } + uint32_t getTransmitAirtime(mesh::Packet*) const { return 100; } + uint32_t calcFloodTimeoutMillisFor(uint32_t) const { return 5000; } + uint32_t calcDirectTimeoutMillisFor(uint32_t, uint8_t) const { return direct_timeout; } + void startConnection(const ContactInfo&, uint16_t) { ++connection_starts; } + int sendLogin(const ContactInfo&, const char*, uint32_t&); + int sendAnonReq(const ContactInfo&, const uint8_t*, uint8_t, uint32_t&, uint32_t&); + int sendRequest(const ContactInfo&, const uint8_t*, uint8_t, uint32_t&, uint32_t&); + int sendRequest(const ContactInfo&, uint8_t, uint32_t&, uint32_t&); +}; + +struct MyMesh : BaseChatMesh { + BaseSerialInterface* _serial = nullptr; + Tracker _delayed_replies; + bool _request_tag_rejected = false; + uint8_t before[16] = {}, out_frame[MAX_FRAME_SIZE + 1] = {}, after[16] = {}; + uint8_t cmd_frame[MAX_FRAME_SIZE + 1] = {}; + ContactInfo recipient; + BaseSerialInterface *private_key_backup_route = nullptr, *command_radio_reply_route = nullptr; + BaseSerialInterface* sign_data_reply_route = nullptr; + uint32_t private_key_backup_deadline = 0; + char private_key_backup_nonce[17] = {}; + uint8_t private_key_backup_sender[6] = {}; + bool command_radio_apply_pending = false; + struct AckBoundary { uint32_t ack = 0; BaseSerialInterface* reply_route = nullptr; }; + AckBoundary expected_ack_table[EXPECTED_ACK_TABLE_SIZE]; + BufferStream terminal_stream; + bool _terminal_login_pending = false, _terminal_trace_pending = false; + uint8_t _terminal_login_key[32] = {}, _terminal_trace_hash_size = 0; + uint8_t _terminal_trace_history = Tracker::NO_HISTORY; + uint32_t _terminal_login_expires_at = 0, _terminal_trace_tag = 0, _terminal_trace_auth = 0; + uint32_t _terminal_trace_sent_at = 0, _terminal_trace_expires_at = 0; + char _terminal_login_target[32] = "synthetic", _terminal_trace_target[32] = "synthetic"; + MyMesh() { + memset(before, 0xa5, sizeof(before)); memset(after, 0xa5, sizeof(after)); + memset(out_frame, 0x5a, sizeof(out_frame)); terminal_stream.write_capacity = 4096; + } + void checkGuards() const { + for (auto byte : before) assert(byte == 0xa5); + for (auto byte : after) assert(byte == 0xa5); + assert(out_frame[MAX_FRAME_SIZE] == 0x5a); + } + ContactInfo* lookupContactByPubKey(const uint8_t* key, size_t len) { + return len == 32 && memcmp(key, recipient.id.pub_key, 32) == 0 ? &recipient : nullptr; + } + bool addContact(const ContactInfo&) { return false; } + bool hasTerminalOutput() const { return true; } + Stream& terminalOutput() { return terminal_stream; } + void cancelPendingRadioParamApply() { command_radio_reply_route = nullptr; } + void cancelSigningSession() { sign_data_reply_route = nullptr; } + void expireExpectedAcks() {} + bool millisHasNowPassed(unsigned long) const; + unsigned long futureMillis(int) const; + void writeErrFrame(uint8_t, BaseSerialInterface* = nullptr); + size_t writePendingSerialFrame(const uint8_t*, size_t, uint32_t); + bool beginPendingRequest(Kind, const ContactInfo&, bool = false); + bool allowRequestTag(uint32_t) override; + void armPendingRequest(uint32_t, uint32_t, bool); + void finishPendingRequest(int, uint32_t, uint32_t); + void abandonPendingRequest(); + void clearPendingReqs(); + bool hasPendingReqs() const; + void servicePendingSerialReply(); + void servicePendingSerialReply(uint32_t); + void clearBinaryTraceReply(); + void serviceBinaryTraceReply(); + void serviceBinaryTraceReply(uint32_t); + void cancelSerialOperationsForRoute(BaseSerialInterface*); + bool hasFiniteDelayedReplyForRoute(BaseSerialInterface*) const; + void onContactResponse(const ContactInfo&, const uint8_t*, uint8_t) override; + bool onContactPathRecv(ContactInfo&, uint8_t*, uint8_t, uint8_t*, uint8_t, + uint8_t, uint8_t*, uint8_t); + void onTraceRecv(mesh::Packet*, uint32_t, uint32_t, uint8_t, + const uint8_t*, const uint8_t*, uint8_t); + void handleRequestFrame(size_t); + void clearTerminalLogin(); + void clearTerminalLogin(uint32_t); + void serviceTerminalLogin(); + void serviceTerminalLogin(uint32_t); + void sendTerminalLogin(ContactInfo&,const char*); + void printTerminalSendStatus(const char*,const ContactInfo&,int,uint32_t); + void clearTerminalTrace(); + void clearTerminalTrace(uint32_t); + void serviceTerminalTrace(); + void serviceTerminalTrace(uint32_t); + void sendTerminalTraceRoute(const uint8_t*, uint8_t, uint8_t, const char*); +}; +#include "production_replies.inc" + +static bool usb_connected = true, other_connected = true; +static bool usbConnected() { return usb_connected; } +static bool otherConnected() { return other_connected; } +struct Fixture { + BufferStream usb_stream, other_stream; + ArduinoSerialInterface usb, other; + SerialWifiInterface tcp; + std::shared_ptr socket; + MultiSerialInterface manager; + BaseSerialInterface* owner = nullptr; + MyMesh mesh; + explicit Fixture(bool tcp_owner = false) { + g_mock_millis = 100; clock_step=0; usb_connected = other_connected = true; + WiFiServer::incoming.clear(); + usb_stream.write_capacity = other_stream.write_capacity = 0; + usb.begin(usb_stream); usb.enableFlowControl(true); usb.setConnectedCheck(usbConnected); + other.begin(other_stream); other.enableFlowControl(true); other.setConnectedCheck(otherConnected); + tcp.begin(5000); + assert(manager.addInterface(InterfaceType::USB, &usb)); + assert(manager.addInterface(InterfaceType::Bluetooth, &other)); + assert(manager.addInterface(InterfaceType::WiFi, &tcp)); + manager.enable(); mesh._serial = &manager; + if (tcp_owner) { + socket = std::make_shared(IPAddress(192, 168, 1, 20)); + socket->received = {'<', 1, 0, 1}; WiFiServer::incoming.push_back(WiFiClient(socket)); owner = &tcp; + uint8_t command[MAX_FRAME_SIZE]; assert(manager.checkRecvFrame(command) == 1); + } else { select(usb_stream, &usb); owner = &usb; } + assert(manager.captureReplyRoute() == owner); + } + ~Fixture() { tcp.end(); } + void select(BufferStream& stream, BaseSerialInterface* expected) { + const uint8_t request[] = {'<', 1, 0, 1}; stream.push(request, sizeof(request)); + uint8_t command[MAX_FRAME_SIZE]; assert(manager.checkRecvFrame(command) == 1); + assert(manager.captureReplyRoute() == expected); + } + void otherRequest() { select(other_stream, &other); } + unsigned fill(BaseSerialInterface* target = nullptr) { + if (!target) target = owner; + const uint8_t required[] = {RESP_CODE_OK, 0x55}; unsigned count = 0; + while (target->writeFrame(required, sizeof(required)) == sizeof(required)) { assert(++count <= 4); } + return count; + } + void service() { mesh.servicePendingSerialReply(); mesh.serviceBinaryTraceReply(); } + void drain() { + usb_stream.write_capacity = other_stream.write_capacity = 4096; + if (socket) socket->write_limit = 4096; + uint8_t command[MAX_FRAME_SIZE]; + for (unsigned i = 0; i != 20; ++i) { manager.loop(); manager.checkRecvFrame(command); service(); } + mesh.checkGuards(); + } + void disconnect() { if (socket) socket->connected = false; else usb_connected = false; manager.loop(); } + const Bytes& wire() const { return socket ? socket->sent : usb_stream.output; } + void arm(Kind kind, uint32_t tag = 17, uint32_t timeout = 5000) { + if (kind == Tracker::Trace) { + assert(mesh._delayed_replies.reserveBinaryTrace(tag, 23, owner, millis())); + mesh._delayed_replies.armBinaryTrace(timeout, millis()); mesh.serviceBinaryTraceReply(); + } else { + assert(mesh.beginPendingRequest(kind, mesh.recipient)); + assert(kind == Tracker::Login || mesh.allowRequestTag(tag)); + mesh.armPendingRequest(tag, timeout, false); + } + } + void command(Kind kind, bool anonymous = false, uint32_t trace_tag = 17) { + memset(mesh.cmd_frame, 0, sizeof(mesh.cmd_frame)); + unsigned offset = 1, len = 33; + switch (kind) { + case Tracker::Login: mesh.cmd_frame[0] = CMD_SEND_LOGIN; len = 34; mesh.cmd_frame[33] = 'x'; break; + case Tracker::Status: mesh.cmd_frame[0] = CMD_SEND_STATUS_REQ; break; + case Tracker::Telemetry: mesh.cmd_frame[0] = CMD_SEND_TELEMETRY_REQ; offset = 4; len = 36; break; + case Tracker::Discovery: mesh.cmd_frame[0] = CMD_SEND_PATH_DISCOVERY_REQ; offset = 2; len = 34; break; + case Tracker::Binary: mesh.cmd_frame[0] = anonymous ? CMD_SEND_ANON_REQ : CMD_SEND_BINARY_REQ; len = 34; break; + case Tracker::Trace: { + mesh.cmd_frame[0] = CMD_SEND_TRACE_PATH; const uint32_t tag = trace_tag, auth = 23; + memcpy(mesh.cmd_frame + 1, &tag, 4); memcpy(mesh.cmd_frame + 5, &auth, 4); + mesh.cmd_frame[10] = 0x12; mesh.handleRequestFrame(11); return; + } + default: assert(false); + } + memcpy(mesh.cmd_frame + offset, mesh.recipient.id.pub_key, 32); + mesh.handleRequestFrame(len); + } +}; + +static const Kind kinds[] = {Tracker::Login, Tracker::Status, Tracker::Telemetry, + Tracker::Binary, Tracker::Discovery, Tracker::Trace}; +static uint8_t code(Kind kind) { + switch (kind) { + case Tracker::Login: return PUSH_CODE_LOGIN_SUCCESS; + case Tracker::Status: return PUSH_CODE_STATUS_RESPONSE; + case Tracker::Telemetry: return PUSH_CODE_TELEMETRY_RESPONSE; + case Tracker::Binary: return PUSH_CODE_BINARY_RESPONSE; + case Tracker::Discovery: return PUSH_CODE_PATH_DISCOVERY_RESPONSE; + case Tracker::Trace: return PUSH_CODE_TRACE_DATA; + default: assert(false); return 0; + } +} +static std::vector frames(const Bytes& wire) { + std::vector result; size_t at = 0; + while (at < wire.size()) { + assert(at + 3 <= wire.size() && wire[at] == '>'); + size_t len = wire[at + 1] + (size_t(wire[at + 2]) << 8); at += 3; + assert(len && at + len <= wire.size()); result.emplace_back(wire.begin() + at, wire.begin() + at + len); at += len; + } + return result; +} +static unsigned count(const Bytes& wire, uint8_t expected) { + unsigned n = 0; for (const auto& frame : frames(wire)) n += frame[0] == expected; return n; +} +static Bytes frameWith(const Bytes& wire, uint8_t expected) { + for (const auto& frame : frames(wire)) if (frame[0] == expected) return frame; + return {}; +} +static void reply(Fixture& fixture, Kind kind, const ContactInfo* from = nullptr, + uint32_t tag = 17, uint8_t marker = 0x65, unsigned length = 64) { + const ContactInfo& contact = from ? *from : fixture.mesh.recipient; + uint8_t data[256] = {}; memcpy(data, &tag, 4); data[4] = marker; + if (kind == Tracker::Trace) { + uint8_t path[] = {marker}; mesh::Packet packet; + fixture.mesh.onTraceRecv(&packet, tag, 23, 0, path, path, 1); + } else if (kind == Tracker::Discovery) { + uint8_t path[] = {marker}; ContactInfo mutable_contact = contact; + fixture.mesh.onContactPathRecv(mutable_contact, path, 1, path, 1, + PAYLOAD_TYPE_RESPONSE, data, 5); + } else { + if (kind == Tracker::Login) { data[4] = RESP_SERVER_LOGIN_OK; data[5] = 1; data[6] = 1; + data[7] = marker; data[12] = 13; length = 13; } + fixture.mesh.onContactResponse(contact, data, uint8_t(length)); + } + fixture.mesh.checkGuards(); +} +static const Tracker::Reply& slot(const Fixture& f, Kind kind) { + return kind == Tracker::Trace ? f.mesh._delayed_replies.trace : f.mesh._delayed_replies.request; +} +static void assertSentBeforeFinal(const Bytes& wire, Kind kind, uint32_t tag = 17) { + bool sent = false; unsigned results = 0; + for (const auto& frame : frames(wire)) { + if (frame[0] == RESP_CODE_SENT) { + assert(!sent && frame.size() == 10); uint32_t actual; memcpy(&actual, frame.data() + 2, 4); + assert(actual == tag); sent = true; + } + if (frame[0] == code(kind)) { assert(sent); ++results; } + } + assert(sent && results == 1); +} + +static void serviceIdleHistory(Tracker& tracker, bool request_service, uint32_t now) { + if (request_service) tracker.serviceRequest(nullptr, now); + else tracker.serviceBinaryTrace(nullptr, now); + assert(tracker.request.phase == Tracker::Empty && tracker.trace.phase == Tracker::Empty); +} + +static void testRetiredHistoryPruning(bool request_service, uint32_t start, + unsigned days, bool daily_service) { + constexpr uint32_t day_ms = 24U * 60U * 60U * 1000U; + const uint8_t login_peer[32] = {1}, status_peer[32] = {2}; + Tracker tracker; + assert(tracker.reserveRequest(Tracker::Login, login_peer, nullptr, true, start)); + tracker.rememberLoginTag(101); + tracker.armRequest(101, 5000, false, start); + tracker.retireRequest(start); + assert(tracker.reserveRequest(Tracker::Status, status_peer, nullptr, true, start)); + assert(tracker.allowRequestTag(17)); + tracker.armRequest(17, 5000, false, start); + tracker.retireRequest(start); + for (unsigned i = 0; i < Tracker::HISTORY_SIZE; ++i) { + const uint8_t history = tracker.reserveTrace(i + 1, 23, start); + assert(history != Tracker::NO_HISTORY); + tracker.retireTrace(history, start, start + 6000U); + } + + // Both Reply slots are empty. Servicing only one slot must still prune both + // histories at the first exact expiry, rather than waiting for a new send. + serviceIdleHistory(tracker, request_service, start + 15999U); + assert(tracker.blocksResponse(login_peer, 999, start + 15999U)); + assert(tracker.blocksResponse(status_peer, 17, start + 15999U)); + for (unsigned i = 0; i < Tracker::HISTORY_SIZE; ++i) + assert(tracker.blocksTrace(i + 1, 23, start + 15999U)); + serviceIdleHistory(tracker, request_service, start + 16000U); + + // Checking a future time is non-mutating. It distinguishes physical pruning + // from merely ignoring an expired guard until the signed range flips again. + const uint32_t far_after_trace_expiry = start + 16000U + 0x80000000U; + assert(!tracker.blocksResponse(status_peer, 17, far_after_trace_expiry)); + for (unsigned i = 0; i < Tracker::HISTORY_SIZE; ++i) + assert(!tracker.blocksTrace(i + 1, 23, far_after_trace_expiry)); + serviceIdleHistory(tracker, request_service, start + 29999U); + assert(tracker.blocksResponse(login_peer, 999, start + 29999U)); + serviceIdleHistory(tracker, request_service, start + Tracker::LOGIN_QUARANTINE_MS); + + if (daily_service) { + for (unsigned day = 1; day <= days; ++day) + serviceIdleHistory(tracker, request_service, start + day * day_ms); + } + const uint32_t later = start + days * day_ms; + assert(days * day_ms > 0x80000000U); + assert(!tracker.blocksResponse(login_peer, 999, later)); + assert(!tracker.blocksResponse(status_peer, 17, later)); + for (unsigned i = 0; i < Tracker::HISTORY_SIZE; ++i) + assert(!tracker.blocksTrace(i + 1, 23, later)); + + // No reservation occurs before the checks above: reserve's own prune cannot + // conceal a missing idle-service prune. All old identities/tuples are reusable. + assert(tracker.reserveRequest(Tracker::Login, login_peer, nullptr, true, later)); + tracker.abandonRequest(); + assert(tracker.reserveRequest(Tracker::Status, status_peer, nullptr, true, later)); + assert(tracker.allowRequestTag(17)); + tracker.abandonRequest(); + for (unsigned i = 0; i < Tracker::HISTORY_SIZE; ++i) + assert(tracker.reserveTrace(i + 1, 23, later) != Tracker::NO_HISTORY); + assert(tracker.reserveTrace(99, 23, later) == Tracker::NO_HISTORY); +} + +static void testReservedHistorySurvivesIdle(bool request_service, uint32_t start, + Kind kind) { + constexpr uint32_t day_ms = 24U * 60U * 60U * 1000U; + const uint8_t peer[32] = {3}; + Tracker tracker; + assert(tracker.reserveRequest(kind, peer, nullptr, true, start)); + if (kind == Tracker::Login) tracker.rememberLoginTag(17); + else assert(tracker.allowRequestTag(17)); + const uint8_t request_history = tracker.request.history; + uint8_t trace_history[Tracker::HISTORY_SIZE]; + for (unsigned i = 0; i < Tracker::HISTORY_SIZE; ++i) { + trace_history[i] = tracker.reserveTrace(i + 1, 23, start); + assert(trace_history[i] != Tracker::NO_HISTORY); + } + const auto service = [&](uint32_t now) { + if (request_service) tracker.serviceRequest(nullptr, now); + else tracker.serviceBinaryTrace(nullptr, now); + assert(tracker.request.phase == Tracker::Reserved); + assert(tracker.request.history == request_history && tracker.request.kind == kind); + assert(tracker.requestPeer() != nullptr && memcmp(tracker.requestPeer(), peer, 32) == 0); + assert(tracker.blocksResponse(peer, 17, now)); + assert(tracker.trace.phase == Tracker::Empty); + }; + service(start + 16000U); + service(start + Tracker::LOGIN_QUARANTINE_MS); + for (unsigned day = 1; day <= 40; ++day) service(start + day * day_ms); + const uint32_t later = start + 40U * day_ms; + // Reserved entries have no expiry authority, even if stale fields look due. + for (unsigned i = 0; i < Tracker::HISTORY_SIZE; ++i) + assert(tracker.reserveTrace(i + 1, 23, later) == Tracker::NO_HISTORY); + assert(tracker.reserveTrace(99, 23, later) == Tracker::NO_HISTORY); + assert(!tracker.reserveRequest(kind, peer, nullptr, true, later)); + for (unsigned i = 0; i < Tracker::HISTORY_SIZE; ++i) tracker.abandonTrace(trace_history[i]); + tracker.abandonRequest(); + assert(tracker.reserveRequest(kind, peer, nullptr, true, later)); + tracker.abandonRequest(); + for (unsigned i = 0; i < Tracker::HISTORY_SIZE; ++i) + assert(tracker.reserveTrace(i + 1, 23, later) != Tracker::NO_HISTORY); +} + +int main() { + unsigned checks = 0; + for (bool tcp : {false, true}) for (Kind kind : kinds) { + Fixture f(tcp); f.arm(kind); assert(f.fill() == 3); f.otherRequest(); + reply(f, kind); const auto& held = slot(f, kind); + assert(held.phase == Tracker::AwaitAdmission && held.route == f.owner && held.frame_len > 0); + const Bytes original(held.frame, held.frame + held.frame_len); const uint32_t deadline = held.delivery_deadline; + assert(f.mesh.hasFiniteDelayedReplyForRoute(f.owner)); + g_mock_millis += 10; reply(f, kind, nullptr, 17, 0x77); + assert(slot(f, kind).delivery_deadline == deadline); + assert(Bytes(slot(f, kind).frame, slot(f, kind).frame + slot(f, kind).frame_len) == original); + assert(f.mesh.connection_starts == (kind == Tracker::Login ? 1U : 0U)); + f.drain(); assertSentBeforeFinal(f.wire(), kind); assert(frameWith(f.wire(), code(kind)) == original); + assert(f.other_stream.output.empty() && slot(f, kind).phase == Tracker::Empty); ++checks; + } + for (bool tcp : {false, true}) for (Kind kind : kinds) { + Fixture f(tcp); assert(f.fill() == 4); f.arm(kind); assert(slot(f, kind).sent_pending); + const uint32_t first_deadline = slot(f, kind).sent_deadline; + g_mock_millis += 1000; f.otherRequest(); reply(f, kind); + assert(slot(f, kind).delivery_deadline == first_deadline); + f.drain(); assertSentBeforeFinal(f.wire(), kind); assert(f.other_stream.output.empty()); ++checks; + } + for (bool tcp : {false, true}) for (Kind kind : kinds) for (unsigned reason = 0; reason < 4; ++reason) { + Fixture f(tcp); f.arm(kind); + if (reason == 0) f.mesh.cancelSerialOperationsForRoute(f.owner); + else if (reason == 1) { f.disconnect(); f.service(); } + else { g_mock_millis = slot(f, kind).radio_deadline + (reason == 3); } + f.manager.forgetReplyRouteForDisconnected(f.owner); f.otherRequest(); reply(f, kind); f.drain(); + assert(slot(f, kind).phase == Tracker::Empty); + assert(count(f.wire(), code(kind)) == 0 && f.other_stream.output.empty()); ++checks; + } + for (bool tcp : {false, true}) for (Kind kind : kinds) { + Fixture f(tcp); f.arm(kind); reply(f, kind); f.otherRequest(); reply(f, kind); f.drain(); + assertSentBeforeFinal(f.wire(), kind); assert(f.other_stream.output.empty()); ++checks; + } + for (bool tcp : {false, true}) for (Kind kind : {Tracker::Status, Tracker::Telemetry, Tracker::Binary, Tracker::Discovery}) { + Fixture f(tcp); f.arm(kind); ContactInfo wrong = f.mesh.recipient; wrong.id.pub_key[0] ^= 0x40; + f.otherRequest(); reply(f, kind, &wrong); assert(slot(f, kind).phase == Tracker::AwaitRadio); + assert(f.other_stream.output.empty()); reply(f, kind); f.drain(); assertSentBeforeFinal(f.wire(), kind); ++checks; + } + for (bool tcp : {false, true}) { + Fixture f(tcp); f.arm(Tracker::Login); ContactInfo collision = f.mesh.recipient; collision.id.pub_key[12] ^= 0x80; + reply(f, Tracker::Login, &collision); assert(f.mesh._delayed_replies.request.phase == Tracker::AwaitRadio); + assert(f.mesh.connection_starts == 0); reply(f, Tracker::Login); f.drain(); + assert(count(f.wire(), PUSH_CODE_LOGIN_SUCCESS) == 1 && f.mesh.connection_starts == 1); ++checks; + } + for (Kind kind : {Tracker::Status, Tracker::Telemetry, Tracker::Binary}) { + Fixture f; f.arm(kind); reply(f, kind, nullptr, 17, 0x65, 4); + assert(f.mesh._delayed_replies.request.phase == Tracker::AwaitRadio); + reply(f, kind, nullptr, 17, 0x65, 255); assert(f.mesh._delayed_replies.request.phase == Tracker::AwaitRadio); + reply(f, kind); f.drain(); assertSentBeforeFinal(f.wire(), kind); ++checks; + } + { + Fixture f; f.arm(Tracker::Discovery); uint8_t path[64] = {}, extra[5] = {}; uint32_t tag = 17; + memcpy(extra, &tag, 4); f.mesh.onContactPathRecv(f.mesh.recipient, path, 0xc0, path, 1, + PAYLOAD_TYPE_RESPONSE, extra, sizeof(extra)); + assert(f.mesh._delayed_replies.request.phase == Tracker::AwaitRadio); + reply(f, Tracker::Discovery); f.drain(); assertSentBeforeFinal(f.wire(), Tracker::Discovery); ++checks; + } + { + Fixture f; f.arm(Tracker::Trace); mesh::Packet packet; uint8_t data[256] = {}; + f.mesh.onTraceRecv(&packet, 17, 23, 0, data, data, 255); + assert(f.mesh._delayed_replies.trace.phase == Tracker::AwaitRadio); + reply(f, Tracker::Trace); f.drain(); assertSentBeforeFinal(f.wire(), Tracker::Trace); ++checks; + } + for (uint32_t start : {100U, 0xfffffff0U}) for (Kind kind : kinds) { + Fixture f; g_mock_millis = start; f.arm(kind); assert(f.fill() == 3); reply(f, kind); + const uint32_t expiry = slot(f, kind).delivery_deadline; g_mock_millis = expiry - 1; + f.service(); assert(slot(f, kind).phase == Tracker::AwaitAdmission); + g_mock_millis = expiry; f.service(); assert(slot(f, kind).phase == Tracker::Empty); + f.otherRequest(); reply(f, kind); f.drain(); assert(count(f.wire(), code(kind)) == 0); + assert(f.other_stream.output.empty()); ++checks; + } + { + Fixture f; f.arm(Tracker::Binary); assert(f.fill() == 3); reply(f, Tracker::Binary); + f.otherRequest(); assert(f.mesh._delayed_replies.reserveBinaryTrace(18, 23, &f.other, millis())); + f.mesh._delayed_replies.armBinaryTrace(5000, millis()); f.mesh.serviceBinaryTraceReply(); + assert(f.fill(&f.other) == 3); reply(f, Tracker::Trace, nullptr, 18); + assert(f.mesh._delayed_replies.request.phase == Tracker::AwaitAdmission); + assert(f.mesh._delayed_replies.trace.phase == Tracker::AwaitAdmission); + f.mesh.cancelSerialOperationsForRoute(f.owner); + assert(f.mesh._delayed_replies.request.phase == Tracker::Empty); + assert(f.mesh._delayed_replies.trace.phase == Tracker::AwaitAdmission); + f.drain(); assert(count(f.wire(), PUSH_CODE_BINARY_RESPONSE) == 0); + assertSentBeforeFinal(f.other_stream.output, Tracker::Trace, 18); ++checks; + } + { + Fixture f; f.arm(Tracker::Status); uint8_t extra[5] = {}, path[] = {1}; + f.mesh.onContactPathRecv(f.mesh.recipient, path, 1, path, 1, PAYLOAD_TYPE_RESPONSE, extra, 5); + assert(f.mesh._delayed_replies.request.phase == Tracker::AwaitRadio && f.mesh._delayed_replies.request.tag == 17); + reply(f, Tracker::Status); f.drain(); assertSentBeforeFinal(f.wire(), Tracker::Status); ++checks; + } + // A retired login peer is intentionally quarantined because its wire reply + // has no reflected nonce. Other peers' legitimate unsolicited data survives. + { + Fixture f; f.arm(Tracker::Login); f.mesh.cancelSerialOperationsForRoute(f.owner); f.otherRequest(); + assert(!f.mesh.beginPendingRequest(Tracker::Login, f.mesh.recipient)); + reply(f, Tracker::Login, nullptr, 0x11223344); ContactInfo unrelated = f.mesh.recipient; + unrelated.id.pub_key[0] ^= 0x40; reply(f, Tracker::Telemetry, &unrelated, 99); f.drain(); + assert(count(f.other_stream.output, PUSH_CODE_LOGIN_SUCCESS) == 0); + assert(count(f.other_stream.output, PUSH_CODE_BINARY_RESPONSE) == 1); ++checks; + } + // Known reflected old replies must not become a newer login result; and a + // positively matched newer status must remain valid after login quarantine. + { + Fixture f; f.arm(Tracker::Status); reply(f, Tracker::Status); f.drain(); + assert(f.mesh.beginPendingRequest(Tracker::Login, f.mesh.recipient)); f.mesh.armPendingRequest(23, 5000, false); + reply(f, Tracker::Status); assert(f.mesh._delayed_replies.request.phase == Tracker::AwaitRadio); + reply(f, Tracker::Login, nullptr, 0x11223344); f.drain(); assert(count(f.wire(), PUSH_CODE_LOGIN_SUCCESS) == 1); ++checks; + } + { + Fixture f; f.arm(Tracker::Login); reply(f, Tracker::Login, nullptr, 0x11223344); f.drain(); + f.arm(Tracker::Status, 18); reply(f, Tracker::Status, nullptr, 18); f.drain(); + assert(count(f.wire(), PUSH_CODE_STATUS_RESPONSE) == 1); ++checks; + } + // Real Base send methods and real command branches prove rejection happens + // before packet allocation and transmission, including RTC tag re-use. + for (Kind kind : {Tracker::Login, Tracker::Status, Tracker::Telemetry, Tracker::Binary, Tracker::Discovery, Tracker::Trace}) { + Fixture f; f.command(kind); assert(f.mesh.transmissions == 1 && f.mesh.allocations == 1); + f.command(kind); assert(f.mesh.transmissions == 1 && f.mesh.allocations == 1); ++checks; + } + for (bool tcp : {false,true}) for (Kind kind : kinds) for (bool flood : {false,true}) { + Fixture f(tcp); if(flood && kind!=Tracker::Trace) f.mesh.recipient.out_path_len=OUT_PATH_UNKNOWN; + f.command(kind); const auto& active=slot(f,kind);const uint32_t tag=active.tag; + if(kind==Tracker::Login) { uint32_t expected;memcpy(&expected,f.mesh.recipient.id.pub_key,4);assert(tag==expected); } + reply(f,kind,nullptr,kind==Tracker::Login ? 0x11223344U : tag);f.drain(); + assertSentBeforeFinal(f.wire(),kind,tag);const Bytes sent=frameWith(f.wire(),RESP_CODE_SENT); + uint32_t timeout;memcpy(&timeout,sent.data()+6,4);assert(timeout==5000); + assert(sent[1]==uint8_t(kind!=Tracker::Trace && (flood || kind==Tracker::Discovery)));++checks; + } + for(unsigned format=0;format<3;++format) { + Fixture f;f.arm(Tracker::Login);assert(f.fill()==3);uint8_t data[13]={};uint32_t server_time=0x11223344; + memcpy(data,&server_time,4);data[4]=RESP_SERVER_LOGIN_OK;data[5]=1;data[6]=1;data[7]=4;data[12]=13; + f.mesh.onContactResponse(f.mesh.recipient,data,5);assert(slot(f,Tracker::Login).phase==Tracker::AwaitRadio); + f.mesh.onContactResponse(f.mesh.recipient,data,12);assert(slot(f,Tracker::Login).phase==Tracker::AwaitRadio); + data[4]='O';data[5]='K';f.mesh.onContactResponse(f.mesh.recipient,data,5); + assert(slot(f,Tracker::Login).phase==Tracker::AwaitRadio); + unsigned length=13;uint8_t expected=PUSH_CODE_LOGIN_SUCCESS; + if(format==0) length=6; + if(format==1) { data[4]=RESP_SERVER_LOGIN_OK;data[5]=1; } + if(format==2) { data[4]='N';data[5]='O';length=6;expected=PUSH_CODE_LOGIN_FAIL; } + f.mesh.onContactResponse(f.mesh.recipient,data,uint8_t(length)); + assert(slot(f,Tracker::Login).phase==Tracker::AwaitAdmission); + const auto& held=slot(f,Tracker::Login);const Bytes original(held.frame,held.frame+held.frame_len); + data[4]=RESP_SERVER_LOGIN_OK;data[5]=1; + f.mesh.onContactResponse(f.mesh.recipient,data,13); + assert(Bytes(slot(f,Tracker::Login).frame,slot(f,Tracker::Login).frame+slot(f,Tracker::Login).frame_len)==original); + assert(f.mesh.connection_starts==(format==1 ? 1U : 0U));f.drain(); + assert(count(f.wire(),expected)==1 && frameWith(f.wire(),expected)==original);++checks; + } + // The callback captures one time. Crossing a deadline during later reads + // must not invalidate a result which arrived before that deadline. + for(Kind kind : kinds) { + Fixture f;f.arm(kind);assert(f.fill()==3);const uint32_t deadline=slot(f,kind).radio_deadline; + g_mock_millis=deadline-1;clock_step=1;reply(f,kind);clock_step=0; + assert(g_mock_millis==deadline); + assert(slot(f,kind).phase==Tracker::AwaitAdmission);f.drain();assertSentBeforeFinal(f.wire(),kind);++checks; + } + { + Fixture f;f.arm(Tracker::Binary);assert(f.fill()==3); + g_mock_millis=slot(f,Tracker::Binary).radio_deadline-1;reply(f,Tracker::Binary); + g_mock_millis+=1000;f.drain();assertSentBeforeFinal(f.wire(),Tracker::Binary);++checks; + } + { + Fixture f;assert(f.fill()==4);f.arm(Tracker::Binary,17,20000); + const uint32_t host_deadline=slot(f,Tracker::Binary).sent_deadline; + g_mock_millis=host_deadline-1;reply(f,Tracker::Binary); + assert(slot(f,Tracker::Binary).delivery_deadline==host_deadline); + g_mock_millis=host_deadline;f.service();assert(slot(f,Tracker::Binary).phase==Tracker::Empty); + f.drain();assert(count(f.wire(),RESP_CODE_SENT)==0 && count(f.wire(),PUSH_CODE_BINARY_RESPONSE)==0);++checks; + } + for(Kind kind : {Tracker::Status,Tracker::Binary,Tracker::Telemetry,Tracker::Discovery,Tracker::Login,Tracker::Trace}) { + Fixture f;f.mesh.radio_accepts=false; + for(unsigned attempt=0;attempt synthetic peer accepted")!=std::string::npos);++checks; + } + for(bool same_peer : {false,true}) { + Fixture f;f.arm(Tracker::Login);f.mesh._terminal_login_pending=true; + memcpy(f.mesh._terminal_login_key,f.mesh.recipient.id.pub_key,32); + if(!same_peer) f.mesh._terminal_login_key[12]^=0x80; + f.mesh.clearTerminalLogin();assert(f.mesh.hasPendingReqs()); + assert(f.mesh._delayed_replies.request.route==f.owner && !f.mesh._delayed_replies.request.terminal); + reply(f,Tracker::Login);f.drain();assertSentBeforeFinal(f.wire(),Tracker::Login);++checks; + } + { + Fixture f;f.mesh.radio_accepts=false; + for(unsigned attempt=0;attempt0 ? write((const uint8_t*)text, (size_t)n #include #include +#include #define MESH_DEBUG_PRINTLN(...) ((void)0) #define PAYLOAD_TYPE_RESPONSE 1 #define MAX_PATH_SIZE 64 -#define MAX_FRAME_SIZE 176 @CODES@ struct ContactInfo { struct { uint8_t pub_key[32] = {1, 2, 3, 4, 5, 6}; } id; }; -struct BaseSerialInterface {}; struct Clock { uint32_t now = 100; uint32_t getMillis() const { return now; } }; -struct SerialBoundary { +struct SerialBoundary : BaseSerialInterface { bool available = true; unsigned writes = 0; BaseSerialInterface* written_route = nullptr; std::vector written; - bool isReplyRouteAvailable(BaseSerialInterface* route) const { + void enable() override {} + void disable() override {} + bool isEnabled() const override { return true; } + bool isConnected() const override { return available; } + bool isReadBusy() const override { return false; } + bool isWriteBusy() const override { return false; } + size_t checkRecvFrame(uint8_t*) override { return 0; } + size_t writeFrame(const uint8_t* frame, size_t len) override { return writeFrameToRoute(this,frame,len); } + bool isReplyRouteAvailable(BaseSerialInterface* route) const override { return available && route != nullptr; } - size_t writeFrameToRoute(BaseSerialInterface* route, const uint8_t* frame, size_t len) { + size_t writeFrameToRoute(BaseSerialInterface* route, const uint8_t* frame, size_t len) override { ++writes; written_route = route; written.assign(frame, frame + len); return len; } }; @@ -57,19 +64,26 @@ struct BaseChatMesh { } }; struct MyMesh : BaseChatMesh { - uint32_t pending_login = 0, pending_status = 0, pending_telemetry = 0; - uint32_t pending_discovery = 0, pending_req = 0; - BaseSerialInterface* pending_serial_reply_route = nullptr; - unsigned long pending_serial_reply_deadline = 0; + mesh::CompanionDelayedReplies _delayed_replies; SerialBoundary serial; SerialBoundary* _serial = &serial; Clock clock; Clock* _ms = &clock; uint8_t out_frame[MAX_FRAME_SIZE] = {}; bool millisHasNowPassed(unsigned long) const; - size_t writePendingSerialFrame(const uint8_t*, size_t); + size_t writePendingSerialFrame(const uint8_t*, size_t, uint32_t); void clearPendingReqs(); void servicePendingSerialReply(); + void servicePendingSerialReply(uint32_t); + void arm(mesh::CompanionDelayedReplies::Kind kind,BaseSerialInterface* route,uint32_t tag,uint32_t deadline) { + ContactInfo contact; + assert(_delayed_replies.reserveRequest(kind,contact.id.pub_key,route,false,clock.now)); + assert(kind==mesh::CompanionDelayedReplies::Login || _delayed_replies.allowRequestTag(tag)); + _delayed_replies.armRequest(tag,1000,false,clock.now); + servicePendingSerialReply(); + _delayed_replies.request.radio_deadline=deadline; + serial.writes=0;serial.written.clear(); + } bool onContactPathRecv(ContactInfo&, uint8_t*, uint8_t, uint8_t*, uint8_t, uint8_t, uint8_t*, uint8_t); }; @@ -84,32 +98,29 @@ static bool response(MyMesh& mesh, uint32_t tag) { } int main() { - BaseSerialInterface original_route, newer_route; + SerialBoundary original_route, newer_route; unsigned checks = 0; // Radio and transport boundaries are mocks; every correlation decision, // output envelope, expiry, and retirement below executes production code. for (unsigned kind = 0; kind != 4; ++kind) { MyMesh mesh; - uint32_t* pending[] = {&mesh.pending_login, &mesh.pending_status, - &mesh.pending_telemetry, &mesh.pending_req}; - *pending[kind] = 0x11223344; - mesh.pending_serial_reply_route = &original_route; - mesh.pending_serial_reply_deadline = 900; + const mesh::CompanionDelayedReplies::Kind kinds[]={mesh::CompanionDelayedReplies::Login, + mesh::CompanionDelayedReplies::Status,mesh::CompanionDelayedReplies::Telemetry, + mesh::CompanionDelayedReplies::Binary}; + mesh.arm(kinds[kind],&original_route,0x11223344,900); memset(mesh.out_frame, 0x5a, sizeof(mesh.out_frame)); assert(response(mesh, 0)); - assert(*pending[kind] == 0x11223344); - assert(mesh.pending_discovery == 0); - assert(mesh.pending_serial_reply_route == &original_route); - assert(mesh.pending_serial_reply_deadline == 900); + assert(mesh._delayed_replies.request.tag == 0x11223344); + assert(mesh._delayed_replies.request.kind == kinds[kind]); + assert(mesh._delayed_replies.request.route == &original_route); + assert(mesh._delayed_replies.request.radio_deadline == 900); assert(mesh.base_path_calls == 1 && mesh.serial.writes == 0); for (uint8_t byte : mesh.out_frame) assert(byte == 0x5a); ++checks; } { MyMesh mesh; - mesh.pending_discovery = 17; - mesh.pending_serial_reply_route = &original_route; - mesh.pending_serial_reply_deadline = 900; + mesh.arm(mesh::CompanionDelayedReplies::Discovery,&original_route,17,900); assert(!response(mesh, 17)); assert(mesh.base_path_calls == 0 && mesh.serial.writes == 1); assert(mesh.serial.written_route == &original_route); @@ -117,35 +128,28 @@ int main() { PUSH_CODE_PATH_DISCOVERY_RESPONSE, 0, 1, 2, 3, 4, 5, 6, 1, 0x31, 2, 0x21, 0x22}; assert(mesh.serial.written == expected); - assert(mesh.pending_discovery == 0 && mesh.pending_serial_reply_route == nullptr); - assert(mesh.pending_serial_reply_deadline == 0); + assert(!mesh._delayed_replies.hasRequest() && mesh._delayed_replies.request.route == nullptr); ++checks; } { MyMesh mesh; - mesh.pending_discovery = 17; - mesh.pending_serial_reply_route = &original_route; - mesh.pending_serial_reply_deadline = 900; + mesh.arm(mesh::CompanionDelayedReplies::Discovery,&original_route,17,900); assert(response(mesh, 18)); - assert(mesh.pending_discovery == 17 && mesh.base_path_calls == 1); - assert(mesh.pending_serial_reply_route == &original_route && mesh.serial.writes == 0); + assert(mesh._delayed_replies.request.tag == 17 && mesh.base_path_calls == 1); + assert(mesh._delayed_replies.request.route == &original_route && mesh.serial.writes == 0); ++checks; } for (uint32_t now : {100U, 101U}) { MyMesh mesh; - mesh.pending_discovery = 17; - mesh.pending_serial_reply_route = &original_route; - mesh.pending_serial_reply_deadline = 100; + mesh.arm(mesh::CompanionDelayedReplies::Discovery,&original_route,17,100); mesh.clock.now = now; mesh.servicePendingSerialReply(); - assert(mesh.pending_discovery == 0 && mesh.pending_serial_reply_route == nullptr); - mesh.pending_status = 23; - mesh.pending_serial_reply_route = &newer_route; - mesh.pending_serial_reply_deadline = 500; + assert(!mesh._delayed_replies.hasRequest() && mesh._delayed_replies.request.route == nullptr); + mesh.arm(mesh::CompanionDelayedReplies::Status,&newer_route,23,500); assert(response(mesh, 0)); - assert(mesh.pending_status == 23 && mesh.base_path_calls == 1); - assert(mesh.pending_serial_reply_route == &newer_route); - assert(mesh.pending_serial_reply_deadline == 500 && mesh.serial.writes == 0); + assert(mesh._delayed_replies.request.tag == 23 && mesh.base_path_calls == 1); + assert(mesh._delayed_replies.request.route == &newer_route); + assert(mesh._delayed_replies.request.radio_deadline == 500 && mesh.serial.writes == 0); ++checks; } assert(checks == 8); @@ -165,6 +169,7 @@ class CompanionDiscoveryCorrelationTests(unittest.TestCase): methods = "\n".join(extract_braced(source, signature) for signature in ( "size_t MyMesh::writePendingSerialFrame(", "void MyMesh::clearPendingReqs(", "void MyMesh::servicePendingSerialReply(", "bool MyMesh::onContactPathRecv(", + "void MyMesh::servicePendingSerialReply(uint32_t now)", )) methods += "\n" + extract_braced( dispatcher, "bool Dispatcher::millisHasNowPassed(").replace("Dispatcher::", "MyMesh::") @@ -182,7 +187,9 @@ class CompanionDiscoveryCorrelationTests(unittest.TestCase): binary = work / "discovery.exe" compiled = subprocess.run([ compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", *SANITIZERS, - str(cpp), "-o", str(binary), + '-isystem',str(ROOT/'test/mocks'),f'-I{ROOT/"src"}', + '-DCOMPANION_FEATURE_TEXT_TERMINAL=0',str(cpp), + str(ROOT/'src/helpers/CompanionDelayedReplies.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) diff --git a/test/test_companion_radio_settings_transaction.py b/test/test_companion_radio_settings_transaction.py index c1ada833f..3c4e27af6 100644 --- a/test/test_companion_radio_settings_transaction.py +++ b/test/test_companion_radio_settings_transaction.py @@ -233,6 +233,7 @@ SLEEP_HARNESS = r''' #include #include #include +#include static uint32_t now_ms=100; struct Serial { bool hasPendingIO() const { return false; } @@ -244,11 +245,12 @@ struct { } radio_driver; struct MyMesh { bool _radio_available=true, _iter_started=false, command_radio_apply_pending=false; - bool binary_trace_pending=false, saved_radio_apply_pending=false; + bool saved_radio_apply_pending=false, _terminal_trace_pending=false; uint32_t radio_apply_retry_at=0, _scheduled_reboot_at=0; uint32_t emergency_client_repeat_send_at=0, dirty_contacts_expiry=0; uint8_t dirty_contacts_failures=0; - const void *pending_serial_reply_route=nullptr, *sign_data=nullptr; + mesh::CompanionDelayedReplies _delayed_replies; + const void *sign_data=nullptr; const void *emergency_client_repeat_packet=nullptr; Serial* _serial=nullptr; #if COMPANION_FEATURE_TEMP_RADIO @@ -259,6 +261,7 @@ struct MyMesh { bool isDualRadioActive() const { return false; } bool isContactWriteDue() const { return false; } bool hasPendingOtaApply() const { return false; } + bool hasPendingReqs() const { return _delayed_replies.hasRequest(); } bool hasQueuedWorkDue() const { return false; } bool hasRetryWorkDue() const { return false; } bool hasPendingWork() const; @@ -296,6 +299,12 @@ int main() { #endif m.saved_radio_apply_pending=false; assert(!m.hasPendingWork()); + // Terminal TRACE has no binary slot but must keep event-driven nodes awake + // until its independent timeout releases the reserved history entry. + m._terminal_trace_pending=true; + assert(m.hasPendingWork()==bool(COMPANION_FEATURE_TEXT_TERMINAL)); + m._terminal_trace_pending=false; + assert(!m.hasPendingWork()); } ''' @@ -312,11 +321,17 @@ class CompanionRadioSettingsTransactionTests(unittest.TestCase): for macros in (['-DCOMPANION_FEATURE_TEMP_RADIO=0'], ['-DCOMPANION_FEATURE_TEMP_RADIO=1'], ['-DCOMPANION_FEATURE_TEMP_RADIO=1', '-DNRF52_PLATFORM=1']): - with self.subTest(macros=macros): - subprocess.run([os.environ.get('CXX', 'g++'), '-std=c++17', - '-Wall', '-Wextra', '-Werror', *macros, - str(cpp), '-o', str(binary)], check=True) - subprocess.run([str(binary)], check=True) + for terminal in (0, 1): + with self.subTest(macros=macros, terminal=terminal): + subprocess.run([os.environ.get('CXX', 'g++'), '-std=c++17', + '-Wall', '-Wextra', '-Werror', *macros, + '-DCOMPANION_FEATURE_TEXT_TERMINAL=' + str(terminal), + '-fsanitize=address,undefined', '-fno-sanitize-recover=all', + '-fno-pie', '-no-pie', + '-isystem', str(ROOT / "test/mocks"), f'-I{ROOT / "src"}', + str(cpp), str(ROOT / 'src/helpers/CompanionDelayedReplies.cpp'), + '-o', str(binary)], check=True) + subprocess.run([str(binary)], check=True) def test_production_commands_commit_and_restore(self): text = SOURCE.read_text(encoding='utf-8') diff --git a/test/test_companion_response_bounds.py b/test/test_companion_response_bounds.py index 325f6fdfc..01608d348 100644 --- a/test/test_companion_response_bounds.py +++ b/test/test_companion_response_bounds.py @@ -18,6 +18,7 @@ HARNESS = r''' #include #include #include +#include #define COMPANION_FEATURE_TEXT_TERMINAL 0 #define MESH_DEBUG_PRINTLN(...) ((void)0) #define RESP_SERVER_LOGIN_OK 0 @@ -31,36 +32,56 @@ struct Packet { int getRSSI() const { return 0; } }; } -struct Serial { +struct Serial : BaseSerialInterface { std::vector> frames; bool isConnected() const { return true; } - void writeFrame(const uint8_t* data, size_t n) { + void enable() override {} + void disable() override {} + bool isEnabled() const override { return true; } + bool isReadBusy() const override { return false; } + bool isWriteBusy() const override { return false; } + size_t checkRecvFrame(uint8_t*) override { return 0; } + size_t writeFrame(const uint8_t* data, size_t n) override { assert(n <= MAX_FRAME_SIZE); frames.emplace_back(data, data+n); + return n; } - void writeFrameToRoute(Serial*, const uint8_t* data, size_t n) { writeFrame(data,n); } }; +struct Clock { uint32_t getMillis() const { return 100; } }; struct MyMesh { - uint32_t pending_login=0, pending_status=0, pending_telemetry=0, pending_req=0; + mesh::CompanionDelayedReplies _delayed_replies; + Clock clock; + Clock* _ms=&clock; uint8_t before[16], out_frame[MAX_FRAME_SIZE+1], after[16]; Serial serial; Serial* _serial = &serial; - unsigned clears = 0; - bool binary_trace_pending = true; - uint32_t binary_trace_tag = 17, binary_trace_auth = 23; - Serial* binary_trace_reply_route = &serial; MyMesh() { memset(before, 0xA5, sizeof(before)); memset(after, 0xA5, sizeof(after)); memset(out_frame, 0x5A, sizeof(out_frame)); } - void clearPendingReqs() { ++clears;pending_login=pending_status=pending_telemetry=pending_req=0; } + void arm(mesh::CompanionDelayedReplies::Kind kind, const ContactInfo& contact) { + assert(_delayed_replies.reserveRequest(kind, contact.id.pub_key, &serial, false, 100)); + assert(kind==mesh::CompanionDelayedReplies::Login || _delayed_replies.allowRequestTag(17)); + _delayed_replies.armRequest(17,1000,false,100); + servicePendingSerialReply();serial.frames.clear(); + } + void armTrace() { + assert(_delayed_replies.reserveBinaryTrace(17,23,&serial,100)); + _delayed_replies.armBinaryTrace(1000,100); + serviceBinaryTraceReply();serial.frames.clear(); + } + void clearPendingReqs(); void startConnection(const ContactInfo&, uint16_t) {} - void writePendingSerialFrame(const uint8_t* p, size_t n) { serial.writeFrame(p,n); } + size_t writePendingSerialFrame(const uint8_t*, size_t, uint32_t); + void servicePendingSerialReply(); + void servicePendingSerialReply(uint32_t); + void serviceBinaryTraceReply(); + void serviceBinaryTraceReply(uint32_t); void onContactResponse(const ContactInfo&, const uint8_t*, uint8_t); void onControlDataRecv(mesh::Packet*); void onRawDataRecv(mesh::Packet*); void onTraceRecv(mesh::Packet*,uint32_t,uint32_t,uint8_t,const uint8_t*,const uint8_t*,uint8_t); - void clearBinaryTraceReply() { binary_trace_pending = false; } + void clearBinaryTraceReply(); void checkGuards() const { for (uint8_t value:before) assert(value==0xA5); for (uint8_t value:after) assert(value==0xA5); @@ -74,22 +95,23 @@ int main() { MyMesh value; uint32_t tag=17; uint8_t data[256]={};memcpy(data,&tag,4); - if (kind==0) value.pending_status=tag; - if (kind==1) value.pending_telemetry=tag; - if (kind==2) value.pending_req=tag; + const mesh::CompanionDelayedReplies::Kind kinds[]={mesh::CompanionDelayedReplies::Status, + mesh::CompanionDelayedReplies::Telemetry,mesh::CompanionDelayedReplies::Binary}; + value.arm(kinds[kind],contact); value.onContactResponse(contact,data,n); const unsigned minimum=kind==0 ? mesh::COMPANION_MIN_STATUS_RESPONSE_SIZE : 5; const unsigned overhead=kind==2 ? 2 : 4; const bool valid=n>=minimum && n+overhead<=MAX_FRAME_SIZE; assert(value.serial.frames.size()==(valid ? 1U : 0U)); if (valid) assert(value.serial.frames[0].size()==n+overhead); + assert(value._delayed_replies.hasRequest()==!valid); value.checkGuards(); } for (unsigned n=0;n<256;++n) { MyMesh idle;uint8_t data[256]={}; idle.onContactResponse(contact,data,n); const bool unsolicited=n>=4 && n+2<=MAX_FRAME_SIZE; - assert(idle.serial.frames.size()==(unsolicited ? 1U : 0U) && idle.clears==0); + assert(idle.serial.frames.size()==(unsolicited ? 1U : 0U) && !idle._delayed_replies.hasRequest()); if (unsolicited) { const auto& frame=idle.serial.frames[0]; assert(frame.size()==n+2 && frame[0]==PUSH_CODE_BINARY_RESPONSE); @@ -104,18 +126,18 @@ int main() { } } for (bool modern:{false,true}) { - MyMesh value;memcpy(&value.pending_login,contact.id.pub_key,4); + MyMesh value;value.arm(mesh::CompanionDelayedReplies::Login,contact); uint8_t data[13]={};if (!modern) memcpy(data+4,"OK",2); value.onContactResponse(contact,data,modern ? 13 : 6); - assert(value.serial.frames.size()==1 && value.clears==1);value.checkGuards(); + assert(value.serial.frames.size()==1 && !value._delayed_replies.hasRequest());value.checkGuards(); } for (unsigned n=0;n<256;++n) for (uint8_t flags=0;flags<4;++flags) { - MyMesh value;mesh::Packet packet;uint8_t data[256]={}; + MyMesh value;value.armTrace();mesh::Packet packet;uint8_t data[256]={}; value.onTraceRecv(&packet,17,23,flags,data,data,n); const unsigned framed=13+n+(n>>flags); assert(value.serial.frames.size()==(framed<=MAX_FRAME_SIZE ? 1U : 0U)); if (!value.serial.frames.empty()) assert(value.serial.frames[0].size()==framed); - assert(!value.binary_trace_pending);value.checkGuards(); + assert(value._delayed_replies.hasBinaryTrace()==(framed>MAX_FRAME_SIZE));value.checkGuards(); } MyMesh empty;empty.onContactResponse(contact,nullptr,255);empty.checkGuards(); } @@ -131,6 +153,10 @@ class CompanionResponseBoundsTest(unittest.TestCase): methods = '\n'.join(extract_braced(source, signature) for signature in ( 'void MyMesh::onContactResponse(', 'void MyMesh::onControlDataRecv(', 'void MyMesh::onRawDataRecv(', 'void MyMesh::onTraceRecv(', + 'size_t MyMesh::writePendingSerialFrame(', 'void MyMesh::clearPendingReqs(', + 'void MyMesh::servicePendingSerialReply()', 'void MyMesh::servicePendingSerialReply(uint32_t now)', + 'void MyMesh::serviceBinaryTraceReply()', 'void MyMesh::serviceBinaryTraceReply(uint32_t now)', + 'void MyMesh::clearBinaryTraceReply(', )) with tempfile.TemporaryDirectory(prefix='companion-response-') as directory: work = Path(directory) @@ -140,7 +166,9 @@ class CompanionResponseBoundsTest(unittest.TestCase): flags = ['-fsanitize=address,undefined','-fno-sanitize-recover=all','-fno-pie','-no-pie'] if sys.platform.startswith('linux') else [] binary = work / 'test.exe' compiled = subprocess.run([os.environ.get('CXX','g++'),'-std=c++17','-Wall','-Wextra','-Werror', - *flags,f'-I{ROOT / "src"}',str(cpp),'-o',str(binary)], capture_output=True,text=True,timeout=60) + *flags,'-isystem',str(ROOT / "test/mocks"),f'-I{ROOT / "src"}',str(cpp), + str(ROOT / 'src/helpers/CompanionDelayedReplies.cpp'), + '-o',str(binary)], capture_output=True,text=True,timeout=60) self.assertEqual(compiled.returncode,0,compiled.stderr) checked = subprocess.run([str(binary)],capture_output=True,text=True,timeout=10) self.assertEqual(checked.returncode,0,checked.stderr) diff --git a/test/test_companion_wifi_session.py b/test/test_companion_wifi_session.py index 3105173be..fb1de7235 100644 --- a/test/test_companion_wifi_session.py +++ b/test/test_companion_wifi_session.py @@ -3,6 +3,7 @@ from pathlib import Path import os import re import subprocess +import sys import tempfile import unittest @@ -21,6 +22,7 @@ HARNESS = r''' #include #include #include +#include struct FakeInterface : BaseSerialInterface { bool enabled = false; @@ -47,9 +49,8 @@ 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; + mesh::CompanionDelayedReplies _delayed_replies; BaseSerialInterface* command_radio_reply_route = nullptr; - BaseSerialInterface* binary_trace_reply_route = nullptr; BaseSerialInterface* sign_data_reply_route = nullptr; BaseSerialInterface* private_key_backup_route = nullptr; unsigned long private_key_backup_deadline = 0; @@ -61,9 +62,9 @@ struct MyMesh { ++stream_cancels; if (streaming) { streaming = false; interface_manager.unlockReplyRoute(); } } - void clearPendingReqs() { ++pending_cancels; pending_serial_reply_route = nullptr; } + void clearPendingReqs() { ++pending_cancels; _delayed_replies.retireRequest(millis()); } void cancelPendingRadioParamApply() { ++radio_cancels; command_radio_reply_route = nullptr; } - void clearBinaryTraceReply() { ++trace_cancels; binary_trace_reply_route = nullptr; } + void clearBinaryTraceReply() { ++trace_cancels; _delayed_replies.retireBinaryTrace(millis()); } void cancelSigningSession() { ++signing_cancels; sign_data_reply_route = nullptr; } void expireExpectedAcks() { ++expirations; } void cancelSerialOperationsForRoute(BaseSerialInterface*); @@ -85,9 +86,13 @@ int main() { assert(interface_manager.checkRecvFrame(command) == 1); interface_manager.lockReplyRoute(); the_mesh.streaming = true; - the_mesh.pending_serial_reply_route = owner; + const uint8_t peer[32] = {1}; + assert(the_mesh._delayed_replies.reserveRequest( + mesh::CompanionDelayedReplies::Binary, peer, owner, false, millis())); + the_mesh._delayed_replies.armRequest(17, 1000, false, millis()); the_mesh.command_radio_reply_route = owner; - the_mesh.binary_trace_reply_route = owner; + assert(the_mesh._delayed_replies.reserveBinaryTrace(17, 23, owner, millis())); + the_mesh._delayed_replies.armBinaryTrace(1000, millis()); the_mesh.sign_data_reply_route = owner; the_mesh.private_key_backup_route = owner; the_mesh.private_key_backup_deadline = 1234; @@ -108,9 +113,11 @@ int main() { 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._delayed_replies.request.route == (cancelled ? nullptr : owner)); + assert(the_mesh._delayed_replies.hasRequest() == !cancelled); assert(the_mesh.command_radio_reply_route == (cancelled ? nullptr : owner)); - assert(the_mesh.binary_trace_reply_route == (cancelled ? nullptr : owner)); + assert(the_mesh._delayed_replies.trace.route == (cancelled ? nullptr : owner)); + assert(the_mesh._delayed_replies.hasBinaryTrace() == !cancelled); assert(the_mesh.sign_data_reply_route == (cancelled ? nullptr : owner)); assert(the_mesh.private_key_backup_route == (cancelled ? nullptr : owner)); assert(the_mesh.private_key_backup_deadline == (cancelled ? 0UL : 1234UL)); @@ -208,8 +215,11 @@ class CompanionWiFiSessionTest(unittest.TestCase): 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", + *(["-fsanitize=address,undefined", "-fno-sanitize-recover=all", + "-fno-pie", "-no-pie"] if sys.platform.startswith("linux") else []), f"-I{ROOT / 'test/mocks'}", f"-I{ROOT / 'src'}", - "-x", "c++", "-", "-o", str(executable)], + "-x", "c++", "-", str(ROOT / "src/helpers/CompanionDelayedReplies.cpp"), + "-o", str(executable)], input=source, text=True, capture_output=True, ) self.assertEqual(result.returncode, 0, result.stderr) diff --git a/test/test_esp32_usb_serial_hygiene.py b/test/test_esp32_usb_serial_hygiene.py index dfa0b221e..32a06580e 100644 --- a/test/test_esp32_usb_serial_hygiene.py +++ b/test/test_esp32_usb_serial_hygiene.py @@ -79,9 +79,15 @@ class Esp32UsbSerialHygieneTest(unittest.TestCase): mesh = source("examples/companion_radio/MyMesh.cpp") start = mesh.index("bool MyMesh::hasFiniteDelayedReplyForRoute(") body = mesh[start : mesh.index("void MyMesh::servicePendingSerialReply()", start)] - self.assertIn("pending_serial_reply_route == route", body) + self.assertIn("_delayed_replies.hasReplyForRoute(route, _ms->getMillis())", body) self.assertIn("command_radio_reply_route == route", body) - self.assertIn("binary_trace_reply_route == route", body) + delayed = source("src/helpers/CompanionDelayedReplies.cpp") + lease = delayed[delayed.index("bool CompanionDelayedReplies::hasReplyForRoute("):] + self.assertIn("{ &request, &trace }", lease) + self.assertIn("reply->route == route", lease) + self.assertIn("reply->phase >= AwaitRadio", lease) + self.assertIn("reply->delivery_deadline : reply->radio_deadline", lease) + self.assertIn("reply->sent_deadline", lease) self.assertIn("expected_ack_table[i].reply_route == route", body) self.assertNotIn("_iter_started", body) self.assertNotIn("lockReplyRoute", body) diff --git a/test/test_nrf52_usb_logging_contract.py b/test/test_nrf52_usb_logging_contract.py index 579efcf0e..acf431cd7 100644 --- a/test/test_nrf52_usb_logging_contract.py +++ b/test/test_nrf52_usb_logging_contract.py @@ -5,6 +5,8 @@ from pathlib import Path import re import unittest +from test_replay_reset_integration import extract_braced + ROOT = Path(__file__).resolve().parents[1] @@ -546,7 +548,7 @@ class Nrf52UsbLoggingContractTest(unittest.TestCase): mesh_source.index("void MyMesh::handleCmdFrame(") ] self.assertIn( - "pending_serial_reply_route == route", session_operations + "_delayed_replies.request.route == route", session_operations ) self.assertIn( "command_radio_reply_route == route", session_operations @@ -560,21 +562,31 @@ class Nrf52UsbLoggingContractTest(unittest.TestCase): # Every delayed Binary mesh response uses a captured requester. A # missing/disconnected route fails closed instead of broadcasting. - self.assertEqual( - mesh_source.count( - "pending_serial_reply_route = _serial->captureReplyRoute();" - ), - 6, - ) + delayed_source = (ROOT / "src/helpers/CompanionDelayedReplies.cpp").read_text() + begin = extract_braced(mesh_source, "bool MyMesh::beginPendingRequest(") + self.assertIn("_serial->captureReplyRoute()", begin) + self.assertIn("_delayed_replies.reserveRequest(kind, contact.id.pub_key,", begin) + for command, kind in (("LOGIN", "Login"), ("ANON_REQ", "Binary"), + ("STATUS_REQ", "Status"), ("PATH_DISCOVERY_REQ", "Discovery"), + ("TELEMETRY_REQ", "Telemetry"), ("BINARY_REQ", "Binary")): + branch = extract_braced(mesh_source, "} else if (cmd_frame[0] == CMD_SEND_" + command) + self.assertIn(f"beginPendingRequest(mesh::CompanionDelayedReplies::{kind}, *recipient)", branch) + self.assertIn("finishPendingRequest(result, tag, est_timeout);", branch) self.assertGreaterEqual( - mesh_source.count("writePendingSerialFrame(out_frame, i);"), 5 + mesh_source.count("writePendingSerialFrame(out_frame, i, now)"), 5 ) pending_writer = mesh_source[ mesh_source.index("size_t MyMesh::writePendingSerialFrame("): mesh_source.index("void MyMesh::writeDisabledFrame(") ] - self.assertIn("pending_serial_reply_route == NULL", pending_writer) - self.assertIn("writeFrameToRoute(", pending_writer) + self.assertIn("_delayed_replies.storeRequest(frame, len, now)", pending_writer) + self.assertIn("_delayed_replies.serviceRequest(_serial, now);", pending_writer) + reserve = extract_braced(delayed_source, "bool CompanionDelayedReplies::reserveRequest(") + self.assertIn("(!terminal && route == nullptr)", reserve) + service = extract_braced(delayed_source, "bool CompanionDelayedReplies::service(") + self.assertIn("serial->isReplyRouteAvailable(reply.route)", service) + self.assertIn("serial->writeFrameToRoute(reply.route,", service) + self.assertNotRegex(service, r"serial->writeFrame\(") self.assertNotIn("_serial->writeFrame(out_frame, i);", pending_writer) self.assertIn( "command_radio_reply_route = _serial->captureReplyRoute();", @@ -653,26 +665,23 @@ class Nrf52UsbLoggingContractTest(unittest.TestCase): self.assertIn( "starts_long_lived_request && hasPendingReqs()", source ) - self.assertEqual( - source.count( - "pending_serial_reply_deadline =\n" - " futureMillis(est_timeout + est_timeout / 5);" - ), - 6, - ) + self.assertEqual(source.count("finishPendingRequest(result, tag, est_timeout);"), 6) + delayed = (ROOT / "src/helpers/CompanionDelayedReplies.cpp").read_text() + arm = extract_braced(delayed, "void CompanionDelayedReplies::arm(") + self.assertIn("timeout + timeout / 5", arm) + self.assertIn("reply.radio_deadline = now + budget;", arm) + self.assertIn("reply.sent_deadline = now + DELIVERY_GRACE_MS;", arm) self.assertIn("servicePendingSerialReply();", source) self.assertIn( "another Companion request is still pending", source ) - self.assertIn( - "binary_trace_reply_route = _serial->captureReplyRoute();", - source, - ) - self.assertIn( - "_serial->writeFrameToRoute(binary_trace_reply_route,", source - ) - self.assertIn("binary_trace_reply_route == route", source) + trace = extract_braced(source, "} else if (cmd_frame[0] == CMD_SEND_TRACE_PATH") + self.assertIn("_delayed_replies.reserveBinaryTrace(tag, auth,", trace) + self.assertIn("_serial->captureReplyRoute(), _ms->getMillis())", trace) + self.assertIn("_delayed_replies.armBinaryTrace(est_timeout, _ms->getMillis());", trace) + self.assertIn("_delayed_replies.serviceBinaryTrace(_serial, now);", source) + self.assertIn("_delayed_replies.trace.route == route", source) self.assertIn("sign_data_reply_route != signing_route", source) self.assertIn("sign_data_reply_route == route", source) diff --git a/test/test_usb_logging_watchdog_roles.py b/test/test_usb_logging_watchdog_roles.py index 70d2c3342..3ccb91782 100644 --- a/test/test_usb_logging_watchdog_roles.py +++ b/test/test_usb_logging_watchdog_roles.py @@ -15,6 +15,7 @@ COMMON_HARNESS = r''' #include #include #include +#include struct Board { bool updating=false, testing=false; bool isOTAUpdateRunning() const{return updating;} @@ -45,9 +46,9 @@ struct RoleMesh { _scheduled_reboot_at=0; bool outbound=false, temporary=false, ota_apply=false, saved_radio_apply_pending=false, temp_radio_handoff_pending=false, - command_radio_apply_pending=false, binary_trace_pending=false, - _iter_started=false, terminal=true; - void* pending_serial_reply_route=nullptr; + command_radio_apply_pending=false, + _iter_started=false, _terminal_trace_pending=false, terminal=true; + mesh::CompanionDelayedReplies _delayed_replies; void* sign_data=nullptr; Store store; Store* _store=&store; @@ -56,6 +57,7 @@ struct RoleMesh { bool hasOutbound() const{return outbound;} bool isAnyTempRadioActive() const{return temporary;} bool hasPendingOtaApply() const{return ota_apply;} + bool hasPendingReqs() const{return _delayed_replies.hasRequest();} bool isTerminalMode() const{return terminal;} #if MESH_USB_CONSOLE_COOPERATIVE bool functional_output=false; @@ -184,18 +186,31 @@ class UsbLoggingWatchdogRoleTests(unittest.TestCase): callback = method((directory / 'main.cpp').read_text(), 'static bool usbLoggingRecoverySafe(void*)') role_checks = r''' - for(auto member : {&RoleMesh::command_radio_apply_pending, - &RoleMesh::binary_trace_pending}){ - the_mesh.*member=true;assert(!usbLoggingRecoverySafe(nullptr)); - the_mesh.*member=false; + the_mesh.command_radio_apply_pending=true;assert(!usbLoggingRecoverySafe(nullptr)); + the_mesh.command_radio_apply_pending=false; + for(auto slot : {&the_mesh._delayed_replies.request, &the_mesh._delayed_replies.trace}){ + slot->phase=mesh::CompanionDelayedReplies::AwaitRadio; + assert(!usbLoggingRecoverySafe(nullptr)); + slot->phase=mesh::CompanionDelayedReplies::AwaitAdmission; + assert(!usbLoggingRecoverySafe(nullptr)); + slot->phase=mesh::CompanionDelayedReplies::Empty; } + // Terminal TRACE owns only a history reservation, not a binary reply slot. + // Its deadline still must be serviced before USB recovery can reset it. + the_mesh._terminal_trace_pending=true; +#if COMPANION_FEATURE_TEXT_TERMINAL + assert(!the_mesh.canRecoverUsbLogging()); + assert(!usbLoggingRecoverySafe(nullptr)); +#else + assert(the_mesh.canRecoverUsbLogging()); + assert(usbLoggingRecoverySafe(nullptr)); +#endif + the_mesh._terminal_trace_pending=false; the_mesh.store.dirty=true;assert(!usbLoggingRecoverySafe(nullptr)); the_mesh.store.dirty=false; the_mesh._scheduled_reboot_at=1;assert(!usbLoggingRecoverySafe(nullptr)); the_mesh._scheduled_reboot_at=0; int route=1; - the_mesh.pending_serial_reply_route=&route;assert(!usbLoggingRecoverySafe(nullptr)); - the_mesh.pending_serial_reply_route=nullptr; the_mesh.sign_data=&route;assert(!usbLoggingRecoverySafe(nullptr)); the_mesh.sign_data=nullptr; the_mesh.serial.pending=true;assert(!usbLoggingRecoverySafe(nullptr)); @@ -244,9 +259,10 @@ class UsbLoggingWatchdogRoleTests(unittest.TestCase): ''' harness = self._harness('bool canRecoverUsbLogging() const;', accessor, callback, role_checks, '', companion_checks) - for usb, mota in product((0, 1), repeat=2): - with self.subTest(usb=usb, mota=mota): - self._compile_and_run(harness, cooperative=usb, usb=usb, mota=mota) + for usb, mota, terminal in product((0, 1), repeat=3): + with self.subTest(usb=usb, mota=mota, terminal=terminal): + self._compile_and_run(harness, cooperative=usb, usb=usb, mota=mota, + terminal=terminal) def test_every_role_loads_durable_state_and_services_recovery_in_normal_loop(self): for role in ('companion_radio', 'simple_repeater', 'simple_room_server', @@ -273,20 +289,24 @@ class UsbLoggingWatchdogRoleTests(unittest.TestCase): .replace('@COMMAND_VETOES@', commands) .replace('@COMPANION_VETOES@', companion)) - def _compile_and_run(self, harness, cooperative, usb=0, mota=0): + def _compile_and_run(self, harness, cooperative, usb=0, mota=0, terminal=0): with tempfile.TemporaryDirectory() as directory: work = Path(directory) cpp = work / 'roles.cpp' cpp.write_text(harness) binary = work / 'roles' flags = ['-DMESH_USB_CONSOLE_COOPERATIVE=' + str(cooperative), - '-DCOMPANION_FEATURE_USB_MOTA_SOURCE=' + str(mota)] + '-DCOMPANION_FEATURE_USB_MOTA_SOURCE=' + str(mota), + '-DCOMPANION_FEATURE_TEXT_TERMINAL=' + str(terminal)] if usb: flags.append('-DENABLE_USB_INTERFACE=1') built = subprocess.run([os.environ.get('CXX', 'g++'), '-std=c++17', '-Wall', '-Wextra', '-Werror', '-Wno-unused-variable', '-fsanitize=address,undefined', '-fno-sanitize-recover=all', - '-fno-pie', '-no-pie', *flags, str(cpp), '-o', str(binary)], + '-fno-pie', '-no-pie', *flags, + '-isystem', str(ROOT / "test/mocks"), f'-I{ROOT / "src"}', + str(cpp), str(ROOT / 'src/helpers/CompanionDelayedReplies.cpp'), + '-o', str(binary)], capture_output=True, text=True) self.assertEqual(built.returncode, 0, built.stdout + built.stderr) ran = subprocess.run([str(binary)], capture_output=True, text=True)