From 6ac2ec357ae2a0c7cc0ca8889a529acd9600c4fe Mon Sep 17 00:00:00 2001 From: mikecarper Date: Tue, 27 Oct 2026 14:08:50 -0700 Subject: [PATCH] Let a new trace replace the previous attempt on its connection --- docs/companion_radio_full.md | 7 + examples/companion_radio/MyMesh.cpp | 5 - src/helpers/CompanionDelayedReplies.cpp | 24 ++-- src/helpers/CompanionDelayedReplies.h | 3 + .../companion_delayed_reply_delivery/test.cpp | 134 +++++++++++++++++- test/test_companion_delayed_reply_delivery.py | 29 +++- 6 files changed, 180 insertions(+), 22 deletions(-) diff --git a/docs/companion_radio_full.md b/docs/companion_radio_full.md index acd4100ae..d92183025 100644 --- a/docs/companion_radio_full.md +++ b/docs/companion_radio_full.md @@ -708,6 +708,13 @@ 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. +A valid new app trace with a fresh tag/auth tuple replaces the previous trace +on that same connection immediately. Until replacement, the longer radio +timeout remains in effect so a late reply can still be returned. Other +connections cannot cancel the current trace. Duplicate tuples, malformed +paths and a full history leave it intact; late replies to a replaced trace +are suppressed. + 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 diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 592102354..129792294 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -7515,11 +7515,6 @@ void MyMesh::handleCmdFrame(size_t len) { writeErrFrame(ERR_CODE_BAD_STATE); } } else if (cmd_frame[0] == CMD_SEND_TRACE_PATH && len > 10 && len - 10 <= MAX_PACKET_PAYLOAD - 9) { - serviceBinaryTraceReply(); - if (_delayed_replies.hasBinaryTrace()) { - writeErrFrame(ERR_CODE_BAD_STATE); - return; - } uint8_t path_len = len - 10; uint8_t flags = cmd_frame[9]; uint8_t path_sz = flags & 0x03; // NEW v1.11+ diff --git a/src/helpers/CompanionDelayedReplies.cpp b/src/helpers/CompanionDelayedReplies.cpp index 65a5a4796..a96383ae7 100644 --- a/src/helpers/CompanionDelayedReplies.cpp +++ b/src/helpers/CompanionDelayedReplies.cpp @@ -39,7 +39,8 @@ uint32_t CompanionDelayedReplies::radioBudget(const Reply& reply) { uint32_t budget = reply.timeout > maximum - extra ? maximum : reply.timeout + extra; if (!reply.terminal) { - // Official apps start an estimate + 5s timer when they receive SENT. + // Request clients add five seconds to SENT's estimate. Keep this longer + // radio window for traces too; their owner may replace one sooner. const uint32_t host_budget = reply.timeout > maximum - HOST_EXTRA_TIMEOUT_MS ? maximum : reply.timeout + HOST_EXTRA_TIMEOUT_MS; if (host_budget > budget) budget = host_budget; @@ -77,14 +78,18 @@ bool CompanionDelayedReplies::store(Reply& reply, const uint8_t* frame, return true; } +bool CompanionDelayedReplies::expired(const Reply& reply, uint32_t now) { + return due(now, reply.phase == AwaitAdmission + ? reply.delivery_deadline : reply.radio_deadline) + || (reply.sent_pending && due(now, reply.sent_deadline)); +} + 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; + || expired(reply, now)) return true; if (reply.sent_pending) { uint8_t sent[10]; sent[0] = 6; // RESP_CODE_SENT, unchanged Companion wire format. @@ -257,9 +262,14 @@ void CompanionDelayedReplies::retireTrace(uint8_t history, uint32_t now, bool CompanionDelayedReplies::reserveBinaryTrace(uint32_t tag, uint32_t auth, BaseSerialInterface* route, uint32_t now) { - if (hasBinaryTrace() || route == nullptr) return false; + if (route == nullptr) return false; + if (trace.phase >= AwaitRadio && expired(trace, now)) retireBinaryTrace(now); + if (hasBinaryTrace() && trace.route != route) return false; + // Reserve the replacement first. A duplicate/quarantined tuple or full + // history must not cancel the current trace. Retired replies stay isolated. uint8_t history = reserveTrace(tag, auth, now); if (history == NO_HISTORY) return false; + if (hasBinaryTrace()) retireBinaryTrace(now); trace.route = route; trace.tag = tag; trace.auth = auth; @@ -309,9 +319,7 @@ bool CompanionDelayedReplies::hasReplyForRoute(BaseSerialInterface* route, uint3 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; + && !expired(*reply, now)) return true; } return false; } diff --git a/src/helpers/CompanionDelayedReplies.h b/src/helpers/CompanionDelayedReplies.h index 88ea46bd5..4f2cb1ec8 100644 --- a/src/helpers/CompanionDelayedReplies.h +++ b/src/helpers/CompanionDelayedReplies.h @@ -52,6 +52,7 @@ private: static uint32_t retirementDeadline(uint32_t now, uint32_t radio_deadline, uint32_t floor); static uint32_t radioBudget(const Reply& reply); + static bool expired(const Reply& reply, uint32_t now); 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); @@ -78,6 +79,8 @@ public: 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); + // A fresh tuple on the same route replaces its current binary trace. + // Refused reservations leave the current operation intact. bool reserveBinaryTrace(uint32_t tag, uint32_t auth, BaseSerialInterface* route, uint32_t now); void armBinaryTrace(uint32_t timeout, uint32_t now); diff --git a/test/fixtures/companion_delayed_reply_delivery/test.cpp b/test/fixtures/companion_delayed_reply_delivery/test.cpp index bd8b10820..041f34645 100644 --- a/test/fixtures/companion_delayed_reply_delivery/test.cpp +++ b/test/fixtures/companion_delayed_reply_delivery/test.cpp @@ -263,6 +263,12 @@ struct Fixture { for (unsigned i = 0; i != 20; ++i) { manager.loop(); manager.checkRecvFrame(command); service(); } mesh.checkGuards(); } + void drainTransportOnly() { + 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); } + } 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) { @@ -275,7 +281,8 @@ struct Fixture { mesh.armPendingRequest(tag, timeout, false); } } - void command(Kind kind, bool anonymous = false, uint32_t trace_tag = 17) { + void command(Kind kind, bool anonymous = false, uint32_t trace_tag = 17, + uint32_t trace_auth = 23, uint8_t trace_flags = 0) { memset(mesh.cmd_frame, 0, sizeof(mesh.cmd_frame)); unsigned offset = 1, len = 33; switch (kind) { @@ -285,8 +292,9 @@ struct Fixture { 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; + mesh.cmd_frame[0] = CMD_SEND_TRACE_PATH; const uint32_t tag = trace_tag, auth = trace_auth; memcpy(mesh.cmd_frame + 1, &tag, 4); memcpy(mesh.cmd_frame + 5, &auth, 4); + mesh.cmd_frame[9] = trace_flags; mesh.cmd_frame[10] = 0x12; mesh.handleRequestFrame(11); return; } default: assert(false); @@ -334,12 +342,13 @@ static Bytes frameWith(const Bytes& wire, uint8_t expected) { return {}; } static void reply(Fixture& fixture, Kind kind, const ContactInfo* from = nullptr, - uint32_t tag = 17, uint8_t marker = 0x65, unsigned length = 64) { + uint32_t tag = 17, uint8_t marker = 0x65, unsigned length = 64, + uint32_t trace_auth = 23) { 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); + fixture.mesh.onTraceRecv(&packet, tag, trace_auth, 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, @@ -475,6 +484,123 @@ static void testReservedHistorySurvivesIdle(bool request_service, uint32_t start int main() { unsigned checks = 0; + // An unreplaced trace retains the longer radio window even after the app's + // advertised estimate, allowing a late result to remain useful. + for (bool tcp : {false, true}) for (uint32_t start : {100U, 0xfffffff0U}) + for (uint32_t estimate : {2000U, 60000U}) { + Fixture f(tcp); g_mock_millis = start; f.mesh.direct_timeout = estimate; + f.command(Tracker::Trace); f.drain(); + assert(slot(f, Tracker::Trace).radio_deadline == start + std::max(estimate + estimate / 5U, estimate + 5000U)); + g_mock_millis = start + estimate + 1U; + reply(f, Tracker::Trace); f.drain(); + assertSentBeforeFinal(f.wire(), Tracker::Trace); + assert(count(f.wire(), RESP_CODE_ERR) == 0); ++checks; + } + // A fresh trace from the same route supersedes its predecessor immediately, + // including just before the longer deadline and after delayed SENT admission. + for (bool tcp : {false, true}) for (uint32_t start : {100U, 0xfffffff0U}) + for (uint32_t estimate : {2000U, 60000U}) for (bool delayed_sent : {false, true}) + for (bool near_deadline : {false, true}) for (bool fresh_auth : {false, true}) { + Fixture f(tcp); g_mock_millis = start; f.mesh.direct_timeout = estimate; + if (delayed_sent) assert(f.fill() == 4); + f.command(Tracker::Trace, false, 17); + if (delayed_sent) { + assert(slot(f, Tracker::Trace).sent_pending); + g_mock_millis += 9000U; + } + f.drain(); + assert(f.mesh.transmissions == 1 && !slot(f, Tracker::Trace).sent_pending); + const Bytes sent = frameWith(f.wire(), RESP_CODE_SENT); + uint32_t advertised; memcpy(&advertised, sent.data() + 6, 4); + assert(advertised == estimate); + g_mock_millis = near_deadline ? slot(f, Tracker::Trace).radio_deadline - 1U : uint32_t(millis()) + 1U; + const size_t retry_start = f.wire().size(); + const uint32_t tag = fresh_auth ? 17U : 18U, auth = fresh_auth ? 24U : 23U; + f.command(Tracker::Trace, false, tag, auth); + assert(f.mesh.transmissions == 2 && f.mesh.allocations == 2 + && "same-route fresh trace must supersede a pending trace"); + assert(slot(f, Tracker::Trace).tag == tag && slot(f, Tracker::Trace).auth == auth + && slot(f, Tracker::Trace).phase == Tracker::AwaitRadio); + // A late first attempt cannot complete or emit an error for this retry. + reply(f, Tracker::Trace, nullptr, 17); + assert(slot(f, Tracker::Trace).tag == tag && slot(f, Tracker::Trace).auth == auth + && slot(f, Tracker::Trace).phase == Tracker::AwaitRadio); + reply(f, Tracker::Trace, nullptr, tag, 0x65, 64, auth); f.drain(); + const Bytes retry_wire(f.wire().begin() + retry_start, f.wire().end()); + assert(count(retry_wire, RESP_CODE_ERR) == 0); + assertSentBeforeFinal(retry_wire, Tracker::Trace, tag); + // Supersession still quarantines the old tuple; an indistinguishable late + // response cannot become another attempt's successful result. + f.command(Tracker::Trace, false, 17); f.drain(); + assert(f.mesh.transmissions == 2 && f.mesh.allocations == 2); + assert(count(f.wire(), RESP_CODE_ERR) == 1); ++checks; + } + for (bool tcp : {false, true}) for (unsigned refusal : {0U, 1U, 2U, 3U}) { + Fixture f(tcp); f.mesh.direct_timeout = 2000; + f.command(Tracker::Trace); f.drain(); + const uint32_t deadline = slot(f, Tracker::Trace).radio_deadline; + const uint8_t history = slot(f, Tracker::Trace).history; + if (refusal == 0) f.otherRequest(); // A different route cannot cancel the owner. + if (refusal == 3) { + for (unsigned i = 0; i < Tracker::HISTORY_SIZE - 1; ++i) { + const auto held = f.mesh._delayed_replies.reserveTrace(100 + i, 23, millis()); + assert(held != Tracker::NO_HISTORY); + f.mesh._delayed_replies.retireTrace(held, millis(), millis() + 6000U); + } + } + f.command(Tracker::Trace, false, refusal == 1 ? 17U : 18U, 23, refusal == 2 ? 1 : 0); + f.drain(); + assert(f.mesh.transmissions == 1 && f.mesh.allocations == 1); + assert(slot(f, Tracker::Trace).tag == 17 && slot(f, Tracker::Trace).auth == 23 + && slot(f, Tracker::Trace).route == f.owner && slot(f, Tracker::Trace).history == history + && slot(f, Tracker::Trace).radio_deadline == deadline); + const Bytes& error_wire = refusal == 0 ? f.other_stream.output : f.wire(); + assert(frameWith(error_wire, RESP_CODE_ERR) == Bytes({RESP_CODE_ERR, + uint8_t(refusal == 2 ? ERR_CODE_ILLEGAL_ARG : ERR_CODE_BAD_STATE)})); + g_mock_millis += 2001U; reply(f, Tracker::Trace); f.drain(); + assertSentBeforeFinal(f.wire(), Tracker::Trace); + assert(count(f.other_stream.output, PUSH_CODE_TRACE_DATA) == 0); ++checks; + } + // Replacement must not flush the superseded producer's retained SENT or + // result merely because transport capacity returned just before CMD36. + for (bool tcp : {false, true}) for (unsigned held : {0U, 1U, 2U}) { + Fixture f(tcp); + if (held < 2) assert(f.fill() == 4); + f.command(Tracker::Trace); + if (held == 2) { + f.drain(); f.usb_stream.write_capacity = f.other_stream.write_capacity = 0; + if (f.socket) f.socket->write_limit = 0; + assert(f.fill() == 4); + } + if (held != 0) reply(f, Tracker::Trace); + assert(slot(f, Tracker::Trace).sent_pending == (held < 2)); + assert(slot(f, Tracker::Trace).phase == (held == 0 ? Tracker::AwaitRadio : Tracker::AwaitAdmission)); + f.drainTransportOnly(); // Free capacity without servicing the old producer. + const size_t retry_start = f.wire().size(); + f.command(Tracker::Trace, false, 18); + assert(f.mesh.transmissions == 2 && slot(f, Tracker::Trace).tag == 18); + reply(f, Tracker::Trace, nullptr, 17); + assert(slot(f, Tracker::Trace).phase == Tracker::AwaitRadio); + reply(f, Tracker::Trace, nullptr, 18); f.drain(); + const Bytes retry_wire(f.wire().begin() + retry_start, f.wire().end()); + assert(count(retry_wire, RESP_CODE_SENT) == 1); + assert(count(retry_wire, RESP_CODE_ERR) == 0); + assertSentBeforeFinal(retry_wire, Tracker::Trace, 18); ++checks; + } + // A dead owner loses exclusivity at expiry, without admitting its old frame. + for (bool tcp : {false, true}) for (uint32_t start : {100U, 0xfffffff0U}) { + Fixture f(tcp); g_mock_millis = start; + f.command(Tracker::Trace); f.drain(); + g_mock_millis = slot(f, Tracker::Trace).radio_deadline; + f.otherRequest(); f.command(Tracker::Trace, false, 18); + assert(f.mesh.transmissions == 2 && slot(f, Tracker::Trace).route == &f.other); + reply(f, Tracker::Trace, nullptr, 17); + assert(slot(f, Tracker::Trace).tag == 18 && slot(f, Tracker::Trace).phase == Tracker::AwaitRadio); + reply(f, Tracker::Trace, nullptr, 18); f.drain(); + assertSentBeforeFinal(f.other_stream.output, Tracker::Trace, 18); + assert(count(f.wire(), PUSH_CODE_TRACE_DATA) == 0); + assert(count(f.other_stream.output, RESP_CODE_ERR) == 0); ++checks; + } // Exercise the real CMD50 producer and callback with a legacy one-admin ACL // reply after the old estimate + 20% cutoff but inside the app's +5s window. for (bool tcp : {false, true}) for (uint32_t start : {100U, 0xfffffff0U}) { diff --git a/test/test_companion_delayed_reply_delivery.py b/test/test_companion_delayed_reply_delivery.py index 9de14815d..12b94c4cc 100644 --- a/test/test_companion_delayed_reply_delivery.py +++ b/test/test_companion_delayed_reply_delivery.py @@ -119,14 +119,23 @@ def production_delayed_reply_inputs(work): class CompanionDelayedReplyDeliveryTests(unittest.TestCase): - def test_actual_producers_routes_deadlines_and_required_admission(self): + def run_fixture(self, legacy_trace_slot=False): compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++") if compiler is None: self.skipTest("a host C++17 compiler is required") with tempfile.TemporaryDirectory(prefix="meshcore-delayed-replies-") as directory: work = Path(directory) production_delayed_reply_inputs(work) - for terminal in (0, 1): + tracker = ROOT / "src/helpers/CompanionDelayedReplies.cpp" + if legacy_trace_slot: + source = tracker.read_text() + reservation = extract_braced(source, "bool CompanionDelayedReplies::reserveBinaryTrace(") + refusing = reservation.replace("{", "{\n if (hasBinaryTrace()) return false;", 1) + source = source.replace(reservation, refusing, 1).replace( + '#include "CompanionDelayedReplies.h"', '#include ') + tracker = work / "legacy_trace_slot.cpp" + tracker.write_text(source, encoding="ascii") + for terminal in ((0,) if legacy_trace_slot else (0, 1)): with self.subTest(terminal=terminal): binary = work / f"delayed-{terminal}.exe" compiled = subprocess.run([ @@ -135,15 +144,25 @@ class CompanionDelayedReplyDeliveryTests(unittest.TestCase): 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(FIXTURE), str(tracker), 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 legacy_trace_slot: + self.assertNotEqual(checked.returncode, 0, "The old active-trace refusal unexpectedly passed") + self.assertIn("same-route fresh trace must supersede a pending trace", checked.stderr) + else: + self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr) + self.assertIn("PASS: delayed production reply delivery", checked.stdout) + + def test_actual_producers_routes_deadlines_and_required_admission(self): + self.run_fixture() + + def test_old_active_slot_refusal_blocks_same_route_fresh_trace(self): + self.run_fixture(legacy_trace_slot=True) if __name__ == "__main__":