Retain and isolate delayed Companion results under backpressure

This commit is contained in:
mikecarper committed 2026-10-03 23:22:34 -07:00
1 parent 94daaf7e40
commit 2e84512ba4
20 files changed
+1798 -340

No files matched your search

+1
View File
@@ -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
+29
View File
@@ -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,
+225 -203
View File
@@ -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<uint8_t>(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
+18 -13
View File
@@ -42,6 +42,7 @@
#include <RTClib.h>
#include <helpers/ArduinoHelpers.h>
#include <helpers/BaseSerialInterface.h>
#include <helpers/CompanionDelayedReplies.h>
#include <helpers/CompanionMotaControl.h>
#include <helpers/IdentityStore.h>
#include <helpers/LogicalMessageCache.h>
@@ -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<PUB_KEY_SIZE> _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;
+3
View File
@@ -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)
+2
View File
@@ -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);
+303
View File
@@ -0,0 +1,303 @@
#include "CompanionDelayedReplies.h"
#include <string.h>
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
+93
View File
@@ -0,0 +1,93 @@
#pragma once
#include "BaseSerialInterface.h"
#include <stdint.h>
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
+5 -5
View File
@@ -1,5 +1,6 @@
#include <helpers/ArduinoSerialInterface.h>
#include <helpers/MultiSerialInterface.h>
#include <helpers/CompanionDelayedReplies.h>
#include <helpers/wifi/SerialWifiInterface.h>
#include <algorithm>
#include <cassert>
@@ -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;
+760
View File
@@ -0,0 +1,760 @@
#include <algorithm>
#include <array>
#include <cassert>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <deque>
#include <limits>
#include <string>
#include <type_traits>
#include <vector>
#include <helpers/ArduinoSerialInterface.h>
#include <helpers/MultiSerialInterface.h>
#include <helpers/wifi/SerialWifiInterface.h>
#include <helpers/CompanionDelayedReplies.h>
#include <helpers/CompanionStatusResponse.h>
#include <helpers/TracePathHelpers.h>
#include <helpers/TxtDataHelpers.h>
#define MESH_DEBUG_PRINTLN(...) ((void)0)
#include "production_constants.inc"
#include "production_stream.inc"
std::deque<WiFiClient> WiFiServer::incoming;
using Tracker = mesh::CompanionDelayedReplies;
static_assert(std::is_nothrow_default_constructible<Tracker>::value,
"The fixed reply tracker constructor must not add exception cleanup");
using Kind = Tracker::Kind;
using Bytes = std::vector<uint8_t>;
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<MockSocket> 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<MockSocket>(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<Bytes> frames(const Bytes& wire) {
std::vector<Bytes> 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<Tracker::HISTORY_SIZE+2;++attempt) {
f.command(kind);assert(slot(f,kind).phase==Tracker::Empty);f.drain();
}
const unsigned sent=f.mesh.transmissions;f.mesh.radio_accepts=true;f.command(kind);
assert(f.mesh.transmissions==sent+1 && slot(f,kind).phase==Tracker::AwaitRadio);++checks;
}
for(uint32_t now : {100U,0xfffffff0U}) {
Tracker tracker;
for(unsigned i=0;i<Tracker::HISTORY_SIZE;++i) {
uint8_t history=tracker.reserveTrace(i+1,23,now);assert(history!=Tracker::NO_HISTORY);
tracker.retireTrace(history,now,now+6000);
}
assert(tracker.reserveTrace(99,23,now)==Tracker::NO_HISTORY);
assert(tracker.reserveTrace(99,23,now+15999)==Tracker::NO_HISTORY);
assert(tracker.reserveTrace(99,23,now+16000)!=Tracker::NO_HISTORY);++checks;
}
for (bool request_service : {false, true}) for (uint32_t start : {100U, 0xfffffff0U}) {
for (unsigned days : {31U, 40U}) for (bool daily_service : {false, true}) {
testRetiredHistoryPruning(request_service, start, days, daily_service); ++checks;
}
for (Kind kind : {Tracker::Login, Tracker::Status}) {
testReservedHistorySurvivesIdle(request_service, start, kind); ++checks;
}
}
{
Fixture f; f.command(Tracker::Binary, true); assert(f.mesh.transmissions == 1);
const uint32_t tag = f.mesh._delayed_replies.request.tag;
reply(f, Tracker::Binary, nullptr, tag); f.drain();
f.mesh.rtc.resetUniqueTime(tag); f.command(Tracker::Binary, true);
assert(f.mesh.transmissions == 1 && f.mesh.allocations == 1); ++checks;
}
for (Kind kind : {Tracker::Status, Tracker::Binary, Tracker::Telemetry}) {
Fixture f; f.command(kind); const uint32_t tag = f.mesh._delayed_replies.request.tag;
reply(f, kind, nullptr, tag); f.drain(); f.mesh.rtc.resetUniqueTime(tag); f.command(kind);
assert(f.mesh.transmissions == 1 && f.mesh.allocations == 1); ++checks;
}
{
Fixture f;
for (unsigned i = 0; i != Tracker::HISTORY_SIZE; ++i) {
f.command(Tracker::Binary); assert(f.mesh.transmissions == i + 1);
const uint32_t tag = f.mesh._delayed_replies.request.tag;
reply(f, Tracker::Binary, nullptr, tag); f.drain();
}
f.command(Tracker::Binary); assert(f.mesh.transmissions == Tracker::HISTORY_SIZE);
assert(f.mesh.allocations == Tracker::HISTORY_SIZE); ++checks;
}
{
Fixture f;
for(unsigned i=0;i<Tracker::HISTORY_SIZE;++i) {
f.command(Tracker::Trace,false,i+1);assert(f.mesh.transmissions==i+1);
reply(f,Tracker::Trace,nullptr,i+1);f.drain();
}
f.command(Tracker::Trace,false,99);
assert(f.mesh.transmissions==Tracker::HISTORY_SIZE && f.mesh.allocations==Tracker::HISTORY_SIZE);
assert(f.mesh._delayed_replies.trace.phase==Tracker::Empty);++checks;
}
{
Fixture f; f.command(Tracker::Trace); f.mesh.cancelSerialOperationsForRoute(f.owner);
f.otherRequest(); f.command(Tracker::Trace); assert(f.mesh.transmissions == 1 && f.mesh.allocations == 1);
reply(f, Tracker::Trace); f.drain();
assert(count(f.other_stream.output, PUSH_CODE_TRACE_DATA) == 0);
assert(frameWith(f.other_stream.output, RESP_CODE_ERR) == Bytes({RESP_CODE_ERR, ERR_CODE_BAD_STATE})); ++checks;
}
// Tombstones must retain more than the most recent retired reflected tag.
{
Fixture f; f.arm(Tracker::Binary, 17); reply(f, Tracker::Binary); f.drain();
f.arm(Tracker::Binary, 18); reply(f, Tracker::Binary, nullptr, 18); f.drain();
f.otherRequest(); reply(f, Tracker::Binary); reply(f, Tracker::Binary, nullptr, 18); f.drain();
assert(f.other_stream.output.empty()); ++checks;
}
#if COMPANION_FEATURE_TEXT_TERMINAL
// The timeout plus retirement grace must stay in the signed half-range.
// Test the last reachable tripled base estimate and the next estimate,
// including a millis wrap while a maximum-duration trace is pending.
for(uint32_t start : {100U,0xfffffff0U}) {
Fixture f;g_mock_millis=start;uint8_t route[]={0x12};
const uint32_t ceiling=0x7fffffffU-Tracker::DELIVERY_GRACE_MS;
const uint32_t last_base=ceiling/3U, last_timeout=last_base*3U;
f.mesh.direct_timeout=last_base;
f.mesh.sendTerminalTraceRoute(route,1,1,"synthetic");
assert(f.mesh._terminal_trace_pending && f.mesh.transmissions==1 && f.mesh.allocations==1);
assert(f.mesh._terminal_trace_expires_at==start+last_timeout);
g_mock_millis=f.mesh._terminal_trace_expires_at-1;f.mesh.serviceTerminalTrace();
assert(f.mesh._terminal_trace_pending);
g_mock_millis+=1;f.mesh.serviceTerminalTrace();assert(!f.mesh._terminal_trace_pending);
assert(f.mesh._delayed_replies.reserveTrace(17,23,millis())==Tracker::NO_HISTORY);
g_mock_millis+=Tracker::DELIVERY_GRACE_MS;
assert(f.mesh._delayed_replies.reserveTrace(17,23,millis())!=Tracker::NO_HISTORY);++checks;
}
{
Fixture f;uint8_t route[]={0x12};
f.mesh.direct_timeout=(0x7fffffffU-Tracker::DELIVERY_GRACE_MS)/3U+1U;
f.mesh.sendTerminalTraceRoute(route,1,1,"synthetic");
assert(!f.mesh._terminal_trace_pending && f.mesh.transmissions==0 && f.mesh.allocations==0);
assert(f.mesh.rng.calls==0 && f.mesh._terminal_trace_history==Tracker::NO_HISTORY);
const std::string text(f.mesh.terminal_stream.output.begin(),f.mesh.terminal_stream.output.end());
assert(text.find("trace timeout is out of range")!=std::string::npos);
for(unsigned i=0;i<Tracker::HISTORY_SIZE;++i)
assert(f.mesh._delayed_replies.reserveTrace(i+1,23,millis())!=Tracker::NO_HISTORY);
assert(f.mesh._delayed_replies.reserveTrace(99,23,millis())==Tracker::NO_HISTORY);++checks;
}
{
Fixture f;f.mesh.sendTerminalLogin(f.mesh.recipient,"x");
assert(f.mesh._terminal_login_pending && f.mesh.transmissions==1);
assert(f.mesh._delayed_replies.request.terminal && f.mesh._delayed_replies.request.route==nullptr);
reply(f,Tracker::Login,nullptr,0x11223344);reply(f,Tracker::Login,nullptr,0x11223344);
assert(!f.mesh._terminal_login_pending && !f.mesh.hasPendingReqs() && f.mesh.connection_starts==1);
f.drain();assert(f.wire().empty() && f.other_stream.output.empty());
const std::string text(f.mesh.terminal_stream.output.begin(),f.mesh.terminal_stream.output.end());
assert(text.find("LOGIN -> 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;attempt<Tracker::HISTORY_SIZE+2;++attempt) {
f.mesh.sendTerminalLogin(f.mesh.recipient,"x");assert(!f.mesh.hasPendingReqs());
}
f.mesh.radio_accepts=true;f.mesh.sendTerminalLogin(f.mesh.recipient,"x");
assert(f.mesh._terminal_login_pending && f.mesh.hasPendingReqs());++checks;
}
{
Fixture f; uint8_t route[] = {0x12}; f.mesh.sendTerminalTraceRoute(route, 1, 1, "synthetic");
assert(f.mesh._terminal_trace_pending && f.mesh.transmissions == 1);
f.command(Tracker::Trace); assert(f.mesh.transmissions == 1);
mesh::Packet packet; uint8_t path[] = {0x12};
f.mesh.onTraceRecv(&packet, 17, 23, 0, path, path, 1);
assert(!f.mesh._terminal_trace_pending); f.command(Tracker::Trace); assert(f.mesh.transmissions == 1); ++checks;
}
{
Fixture f; f.command(Tracker::Trace); uint8_t route[] = {0x12};
f.mesh.sendTerminalTraceRoute(route, 1, 1, "synthetic"); assert(f.mesh.transmissions == 1);
f.mesh.cancelSerialOperationsForRoute(f.owner); f.mesh.rng.calls = 0;
f.mesh.sendTerminalTraceRoute(route, 1, 1, "synthetic"); assert(f.mesh.transmissions == 1); ++checks;
}
#endif
std::printf("PASS: delayed production reply delivery (%u runtime checks)\n", checks);
}
+1
View File
@@ -272,6 +272,7 @@ class BaseChatMesh : public mesh::Mesh {
public:
int sendLogin(const ContactInfo&, const char*, uint32_t&);
int sendAnonReq(const ContactInfo&, const uint8_t*, uint8_t, uint32_t&, uint32_t&);
virtual bool allowRequestTag(uint32_t) { return true; }
int sendRequest(const ContactInfo&, const uint8_t*, uint8_t, uint32_t&, uint32_t&);
int sendRequest(const ContactInfo&, uint8_t, uint32_t&, uint32_t&);
};
+1
View File
@@ -75,6 +75,7 @@ class CompanionAckDeliveryTests(unittest.TestCase):
f"-I{ROOT / 'src'}", str(FIXTURE),
str(ROOT / "src/helpers/ArduinoSerialInterface.cpp"),
str(ROOT / "src/helpers/wifi/SerialWifiInterface.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)
@@ -0,0 +1,143 @@
#!/usr/bin/env python3
"""Run actual delayed reply producers and bounded UART/TCP admission."""
from pathlib import Path
import os
import re
import shutil
import subprocess
import sys
import tempfile
import unittest
from test_replay_reset_integration import extract_braced
ROOT = Path(__file__).resolve().parents[1]
FIXTURE = ROOT / "test/fixtures/companion_delayed_reply_delivery/test.cpp"
SANITIZERS = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
"-fno-pie", "-no-pie"] if sys.platform.startswith("linux") else [])
def production_methods(source, base, packet, dispatcher):
methods = "\n".join(extract_braced(source, signature) for signature in (
"void MyMesh::writeErrFrame(", "size_t MyMesh::writePendingSerialFrame(",
"void MyMesh::onContactResponse(", "bool MyMesh::onContactPathRecv(",
"void MyMesh::onTraceRecv(", "void MyMesh::clearPendingReqs(",
"bool MyMesh::hasPendingReqs(", "bool MyMesh::beginPendingRequest(",
"bool MyMesh::allowRequestTag(", "void MyMesh::armPendingRequest(",
"void MyMesh::finishPendingRequest(",
"void MyMesh::abandonPendingRequest(", "void MyMesh::servicePendingSerialReply(",
"void MyMesh::servicePendingSerialReply(uint32_t now)",
"void MyMesh::clearBinaryTraceReply(", "void MyMesh::serviceBinaryTraceReply(",
"void MyMesh::serviceBinaryTraceReply(uint32_t now)",
"void MyMesh::cancelSerialOperationsForRoute(",
"bool MyMesh::hasFiniteDelayedReplyForRoute(",
))
terminal = "\n".join(extract_braced(source, signature) for signature in (
"void MyMesh::clearTerminalLogin(", "void MyMesh::serviceTerminalLogin(",
"void MyMesh::clearTerminalLogin(uint32_t now)",
"void MyMesh::sendTerminalLogin(", "void MyMesh::printTerminalSendStatus(",
"void MyMesh::serviceTerminalLogin(uint32_t now)",
"void MyMesh::clearTerminalTrace(", "void MyMesh::serviceTerminalTrace(",
"void MyMesh::clearTerminalTrace(uint32_t now)",
"void MyMesh::serviceTerminalTrace(uint32_t now)",
"void MyMesh::sendTerminalTraceRoute(",
))
methods += "\n#if COMPANION_FEATURE_TEXT_TERMINAL\n" + terminal + "\n#endif\n"
methods += "\n" + "\n".join(extract_braced(base, signature) for signature in (
"bool BaseChatMesh::onContactPathRecv(", "int BaseChatMesh::sendLogin(",
"int BaseChatMesh::sendAnonReq(",
"int BaseChatMesh::sendRequest(const ContactInfo& recipient, const uint8_t*",
"int BaseChatMesh::sendRequest(const ContactInfo& recipient, uint8_t",
))
methods += "\n" + "\n".join(
extract_braced(dispatcher, signature).replace("Dispatcher::", "MyMesh::")
for signature in ("bool Dispatcher::millisHasNowPassed(",
"unsigned long Dispatcher::futureMillis(")
)
methods += "\nnamespace mesh {\n" + "\n".join(extract_braced(packet, signature)
for signature in ("bool Packet::isValidPathLen(", "size_t Packet::writePath(")) + "\n}\n"
# Execute real request guards and command branches, excluding unrelated
# firmware features. Packet allocation/transmission are counted boundaries.
handler = extract_braced(source, "void MyMesh::handleCmdFrame(")
prefix = handler[handler.index("{") + 1:handler.index(" if (cmd_frame[0] == CMD_DEVICE_QUERY")]
branches = []
for command in ("LOGIN", "ANON_REQ", "STATUS_REQ", "PATH_DISCOVERY_REQ",
"TELEMETRY_REQ", "BINARY_REQ", "TRACE_PATH"):
block = extract_braced(source, "} else if (cmd_frame[0] == CMD_SEND_" + command)
branches.append(block.replace("} else if", "if", 1) + "\n")
methods += "\nvoid MyMesh::handleRequestFrame(size_t len) {\n" + prefix + "\n".join(branches) + "\n}\n"
return methods
class CompanionDelayedReplyDeliveryTests(unittest.TestCase):
def test_actual_producers_routes_deadlines_and_required_admission(self):
compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
if compiler is None:
self.skipTest("a host C++17 compiler is required")
source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
base = (ROOT / "src/helpers/BaseChatMesh.cpp").read_text()
packet = (ROOT / "src/Packet.cpp").read_text()
dispatcher = (ROOT / "src/Dispatcher.cpp").read_text()
native = (ROOT / "test/test_serial_mode_switch/test_serial_mode_switch.cpp").read_text()
core = (ROOT / "src/MeshCore.h").read_text()
constants = "\n".join(re.findall(
r"^#define (?:PUSH_CODE_\w+|RESP_CODE_\w+|CMD_SEND_\w+|ERR_CODE_\w+|REQ_TYPE_GET_TELEMETRY_DATA)\s+.+$",
source, re.MULTILINE))
wanted = ("PUB_KEY_SIZE", "MAX_PACKET_PAYLOAD", "MAX_PATH_SIZE", "OUT_PATH_UNKNOWN",
"MSG_SEND_FAILED", "MSG_SEND_SENT_DIRECT", "MSG_SEND_SENT_FLOOD",
"PAYLOAD_TYPE_RESPONSE", "PAYLOAD_TYPE_REQ", "PAYLOAD_TYPE_ANON_REQ",
"PAYLOAD_TYPE_ACK", "REQ_TYPE_GET_STATUS", "REQ_TYPE_GET_TELEMETRY_DATA",
"TELEM_PERM_BASE", "ADV_TYPE_NONE", "ADV_TYPE_ROOM", "RESP_SERVER_LOGIN_OK")
for relative in ("examples/companion_radio/MyMesh.h", "src/MeshCore.h", "src/Packet.h", "src/helpers/BaseChatMesh.h",
"src/helpers/ContactInfo.h", "src/helpers/AdvertDataHelpers.h",
"src/helpers/SensorManager.h"):
text = (ROOT / relative).read_text()
constants += "\n" + "\n".join(re.findall(
r"^#define (?:" + "|".join(wanted) + r")\s+.+$", text, re.MULTILINE))
constants += "\n#define EXPECTED_ACK_TABLE_SIZE 2\n"
clock = "\n".join(extract_braced(core, signature) for signature in (
"uint32_t getCurrentTimeUnique()", "void resetUniqueTime(uint32_t time)"))
# printf is an Arduino Print operation absent from the lean shared
# host mock. Add only this I/O boundary, leaving transport logic real.
stream = (ROOT / "test/mocks/Stream.h").read_text()
stream = "#include <cstdarg>\n#include <cstdio>\n" + stream.replace(
"public:", "public:\n size_t printf(const char* format, ...) {\n"
" char text[256]; va_list args; va_start(args, format);\n"
" int n=vsnprintf(text,sizeof(text),format,args); va_end(args);\n"
" return n>0 ? write((const uint8_t*)text, (size_t)n<sizeof(text) ? (size_t)n : sizeof(text)-1) : 0;\n"
" }", 1)
with tempfile.TemporaryDirectory(prefix="meshcore-delayed-replies-") as directory:
work = Path(directory)
(work / "production_constants.inc").write_text(constants, encoding="ascii")
(work / "production_clock.inc").write_text(clock, encoding="ascii")
(work / "production_stream.inc").write_text(
extract_braced(native, "class BufferStream") + ";\n", encoding="ascii")
(work / "production_replies.inc").write_text(
production_methods(source, base, packet, dispatcher), encoding="ascii")
(work / "Stream.h").write_text(stream, encoding="ascii")
(work / "Arduino.h").write_text(
(ROOT / "test/mocks/Arduino.h").read_text(), encoding="ascii")
for terminal in (0, 1):
with self.subTest(terminal=terminal):
binary = work / f"delayed-{terminal}.exe"
compiled = subprocess.run([
compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
"-Wno-unused-parameter", "-Wno-sign-compare", "-Wno-class-memaccess", *SANITIZERS,
f"-DCOMPANION_FEATURE_TEXT_TERMINAL={terminal}", "-DMESH_ENABLE_ONE_KEY_DM=0",
f"-I{work}", f"-I{ROOT / 'test/mocks'}",
f"-I{ROOT / 'test/fixtures/serial_wifi_sessions/mocks'}", f"-I{ROOT / 'src'}",
str(FIXTURE), str(ROOT / "src/helpers/CompanionDelayedReplies.cpp"),
str(ROOT / "src/helpers/ArduinoSerialInterface.cpp"),
str(ROOT / "src/helpers/wifi/SerialWifiInterface.cpp"), "-o", str(binary),
str(ROOT / "src/helpers/TxtDataHelpers.cpp"),
], 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=20)
self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
self.assertIn("PASS: delayed production reply delivery", checked.stdout)
if __name__ == "__main__":
unittest.main()
+48 -41
View File
@@ -24,23 +24,30 @@ HARNESS = r"""
#include <cstdio>
#include <cstring>
#include <vector>
#include <helpers/CompanionDelayedReplies.h>
#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<uint8_t> 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)
@@ -233,6 +233,7 @@ SLEEP_HARNESS = r'''
#include <cassert>
#include <cstdint>
#include <cstddef>
#include <helpers/CompanionDelayedReplies.h>
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')
+48 -20
View File
@@ -18,6 +18,7 @@ HARNESS = r'''
#include <cstring>
#include <vector>
#include <helpers/CompanionStatusResponse.h>
#include <helpers/CompanionDelayedReplies.h>
#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<std::vector<uint8_t>> 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)
+19 -9
View File
@@ -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 <deque>
#include <vector>
#include <helpers/MultiSerialInterface.h>
#include <helpers/CompanionDelayedReplies.h>
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)
+8 -2
View File
@@ -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)
+34 -25
View File
@@ -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)
+35 -15
View File
@@ -15,6 +15,7 @@ COMMON_HARNESS = r'''
#include <cstddef>
#include <cstdint>
#include <initializer_list>
#include <helpers/CompanionDelayedReplies.h>
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)