From e5706bafa126e64ebf69b47a16cc3279e009644a Mon Sep 17 00:00:00 2001 From: mikecarper Date: Fri, 16 Oct 2026 09:11:19 -0700 Subject: [PATCH] Restart contact streams after storage recovery Release cached path snapshots before resetting contacts while retaining the requesting transport and table revision for a complete stream restart. Keep explicit session cancellation unchanged and avoid copying uncached contact buffers. Cover cached and uncached recovery, transport isolation, empty reloads, backpressure, and the former cancellation as a negative control. --- docs/research/companion_contact_cache.md | 7 +- examples/companion_radio/MyMesh.cpp | 7 +- test/test_companion_contact_sleep_contract.py | 8 +- ...st_companion_contact_stream_concurrency.py | 200 +++++++++++++++++- .../test_companion_contact_stream_contract.py | 20 ++ 5 files changed, 232 insertions(+), 10 deletions(-) diff --git a/docs/research/companion_contact_cache.md b/docs/research/companion_contact_cache.md index 945d7c37b..90998486d 100644 --- a/docs/research/companion_contact_cache.md +++ b/docs/research/companion_contact_cache.md @@ -40,8 +40,11 @@ sending along a different route. Contact I/O failures request bounded recovery between callbacks. A raw cached-path read fault first rechecks the current immutable paths without resetting accepted RAM changes or their save deadline. Persistent faults or uncertain file publication reload a readable live image, -otherwise a readable backup, -and discard only unrecoverable contact data if neither works. nRF52 preserves +otherwise a readable backup, and discard only unrecoverable contact data if +neither works. Recovery releases a pending contact-list snapshot before resetting +the table but preserves its requesting route; the changed table revision starts +a fresh list that can finish normally. Explicit session cancellation still stops +the stream. nRF52 preserves healthy pages when another page is damaged. No archival `.bad` files are kept. If damaged contacts cannot be removed, the node remains usable with volatile contacts; explicit saves still report failure, while reboot/shutdown remain diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 8ea78316a..4697cc05f 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -7526,7 +7526,12 @@ void MyMesh::onContactCacheFlushed() { } bool MyMesh::resetContactsForRecovery() { - stopContactsIterator(); + // Release retained paths before resetting the table. Keep the stream owner + // so its changed table revision restarts a complete list on the next pass. +#if MESH_CONTACT_CACHE + _iter_pending_contact = ContactInfo(); +#endif + _iter_contact_pending = false; resetContacts(); MyMesh::onContactCacheFlushed(); return true; diff --git a/test/test_companion_contact_sleep_contract.py b/test/test_companion_contact_sleep_contract.py index fdc698610..5d311800a 100644 --- a/test/test_companion_contact_sleep_contract.py +++ b/test/test_companion_contact_sleep_contract.py @@ -37,7 +37,13 @@ class CompanionContactSleepContractTest(unittest.TestCase): self.assertIn("|| _store->contactRecoveryPending()", function_body(mesh, "bool MyMesh::hasPendingWork() const")) reset = function_body(mesh, "bool MyMesh::resetContactsForRecovery()") - self.assertLess(reset.index("stopContactsIterator()"), reset.index("resetContacts()")) + self.assertNotIn("stopContactsIterator()", reset) + self.assertLess(reset.index("_iter_pending_contact = ContactInfo()"), + reset.index("resetContacts()")) + self.assertIn("#if MESH_CONTACT_CACHE\n _iter_pending_contact = ContactInfo();\n#endif", + reset) + self.assertLess(reset.index("_iter_contact_pending = false"), + reset.index("resetContacts()")) self.assertIn("MyMesh::onContactCacheFlushed()", reset) clear = function_body(mesh, "void MyMesh::onContactCacheFlushed()") self.assertIn("mesh::resetLazyPersistenceAfterSuccess(dirty_contacts_expiry, dirty_contacts_failures)", clear) diff --git a/test/test_companion_contact_stream_concurrency.py b/test/test_companion_contact_stream_concurrency.py index 9cc4d61eb..5c19a9d1c 100644 --- a/test/test_companion_contact_stream_concurrency.py +++ b/test/test_companion_contact_stream_concurrency.py @@ -21,8 +21,12 @@ HARNESS = r''' #include #include #include +#if MESH_CONTACT_CACHE +#include +#endif #include #include "helpers/ArduinoSerialInterface.h" +#include "helpers/LazyPersistence.h" #include "helpers/MultiSerialInterface.h" #define PUB_KEY_SIZE 32 #define MAX_PATH_SIZE 64 @@ -40,13 +44,31 @@ struct ContactInfo { uint8_t type=1, flags=0, out_path_len=0; char name[32]{}; uint32_t last_advert_timestamp=0, gps_lat=0, gps_lon=0, lastmod=1; - bool copyPathTo(uint8_t* out) const { memset(out,0,MAX_PATH_SIZE); return true; } +#if MESH_CONTACT_CACHE + // Model the production ContactPathRef's shared immutable snapshot lifetime. + std::shared_ptr path_owner; +#else + uint8_t out_path[MAX_PATH_SIZE]{}; +#endif + bool copyPathTo(uint8_t* out) const { +#if MESH_CONTACT_CACHE + memset(out,path_owner ? *path_owner : 0,MAX_PATH_SIZE); +#else + memcpy(out,out_path,MAX_PATH_SIZE); +#endif + return true; + } }; struct ContactsIterator { - unsigned next=0, count=0; + unsigned next=0, count=0, base=0; template bool hasNext(T*,ContactInfo& c) { if(next==count) return false; - c=ContactInfo(); c.id.pub_key[0]=static_cast(++next); + c=ContactInfo(); c.id.pub_key[0]=static_cast(base + ++next); +#if MESH_CONTACT_CACHE + c.path_owner=std::make_shared(c.id.pub_key[0]); +#else + memset(c.out_path,c.id.pub_key[0],MAX_PATH_SIZE); +#endif snprintf(c.name,sizeof(c.name),"HIL-%u",next); c.lastmod=next; return true; } @@ -85,13 +107,27 @@ struct MyMesh { uint32_t _iter_filter_since=0, _iter_next_frame_at=0, _iter_total_count=0; uint32_t _iter_table_revision=0, _most_recent_lastmod=0; bool _iter_started=false, _iter_start_pending=false, _iter_contact_pending=false; - unsigned contacts=54, revision=1; + unsigned contacts=54, revision=1, contact_base=0, recovery_resets=0; + unsigned long dirty_contacts_expiry=123; + uint8_t dirty_contacts_failures=3; uint8_t out_frame[MAX_FRAME_SIZE+4]{},cmd_frame[MAX_FRAME_SIZE+1]{}; bool millisHasNowPassed(uint32_t time) const {return static_cast(time-millis())<0;} uint32_t futureMillis(uint32_t delta) const {return millis()+delta;} uint32_t getContactTableRevision() const {return revision;} unsigned getNumContacts() const {return contacts;} - ContactsIterator startContactsIterator() {return {0,contacts};} + ContactsIterator startContactsIterator() {return {0,contacts,contact_base};} + void resetContacts() { + // The real reset retires the table/path owners and increments revision. + // Cached snapshots must be released before reaching this boundary; an + // uncached value owns only inline bytes, which need not be rewritten. +#if MESH_CONTACT_CACHE + assert(!_iter_pending_contact.path_owner); +#endif + assert(!_iter_contact_pending); + ++recovery_resets; ++revision; contacts=0; + } + void onContactCacheFlushed(); + bool resetContactsForRecovery(); void writeErrFrame(uint8_t error) { uint8_t reply[]={RESP_CODE_ERR,error}; assert(_serial->writeFrame(reply,2)==2); } @@ -163,7 +199,132 @@ static void assertBleSnapshot(const Fixture& f,unsigned contacts=54) { assert(at!=f.ble.delivered.end()); } } + +static void assertLatestSnapshot(const std::vector>& frames, + unsigned contacts,unsigned base) { + auto last=std::find_if(frames.rbegin(),frames.rend(),[](const auto& frame) { + return frame[0]==RESP_CODE_CONTACTS_START; + }); + assert(last!=frames.rend() && last->size()==5); + uint32_t advertised=0; memcpy(&advertised,last->data()+1,4); + assert(advertised==contacts); + unsigned rows=0,ends=0; + for(auto frame=last.base();frame!=frames.end();++frame) { + if((*frame)[0]==RESP_CODE_CONTACT) { + assert((*frame)[1]==base + ++rows); + for(unsigned i=0;i old_path=f.mesh._iter_pending_contact.path_owner; +#endif + BaseSerialInterface* owner=f.mesh._iter_reply_route; + const uint32_t revision=f.mesh._iter_table_revision; + f.sendUsb({CMD_GET_BATT_AND_STORAGE}); f.step(); + assert(f.mesh.resetContactsForRecovery()); + assert(f.mesh._iter_started); // Recovery must not abandon the live requester. + assert(f.mesh._iter_reply_route==owner && owner==&f.ble); +#if MESH_CONTACT_CACHE + assert(old_path.expired()); +#endif + assert(!f.mesh._iter_contact_pending); + assert(f.mesh._iter_table_revision==revision && f.mesh.recovery_resets==1); + assert(f.mesh.dirty_contacts_expiry==0 && f.mesh.dirty_contacts_failures==0); + f.mesh.contacts=3; f.mesh.contact_base=64; // Successful store reload. + f.run(100); // Fresh START cannot yet be admitted. + assert(f.mesh._iter_started && f.mesh._iter_start_pending); + assert(f.mesh._iter_table_revision==f.mesh.revision); + assert(count(f.ble.delivered,RESP_CODE_CONTACTS_START)==1); + // Another host's APP_START must neither cancel nor steal the recovered list. + f.sendUsb({CMD_APP_START,0,0,0,0,0,0,0}); f.step(); + assert(f.mesh._iter_started && f.mesh._iter_reply_route==owner); + assert(count(f.usbFrames(),RESP_CODE_SELF_INFO)==1); + f.ble.capacity=4; + for(unsigned steps=0;count(f.ble.delivered,RESP_CODE_CONTACTS_START)!=2;++steps) { + assert(steps<1000); f.step(); + } + f.ble.capacity=0; f.run(70); // Fresh first CONTACT cannot be admitted. + assert(f.mesh._iter_contact_pending && f.mesh._iter_pending_contact.id.pub_key[0]==65); + assert(count(f.ble.delivered,RESP_CODE_CONTACT)==0); + f.sendUsb({CMD_GET_BATT_AND_STORAGE}); f.step(); + f.ble.capacity=4; + for(unsigned steps=0;count(f.ble.delivered,RESP_CODE_CONTACT)!=3;++steps) { + assert(steps<1000); f.step(); + } + f.ble.capacity=0; f.run(70); // END cannot be admitted either. + assert(f.mesh._iter_started && !f.mesh._iter_contact_pending); + assert(count(f.ble.delivered,RESP_CODE_END_OF_CONTACTS)==0); + f.ble.capacity=4; f.finish(); + assert(count(f.ble.delivered,RESP_CODE_CONTACTS_START)==2); + assertLatestSnapshot(f.ble.delivered,3,64); + assert(count(f.usbFrames(),RESP_CODE_BATT_AND_STORAGE)==2); + assert(count(f.usbFrames(),RESP_CODE_CONTACTS_START)==0); + assert(count(f.usbFrames(),RESP_CODE_CONTACT)==0); + assert(count(f.usbFrames(),RESP_CODE_END_OF_CONTACTS)==0); +} + +static void recovery_keeps_usb_owner_including_an_empty_reload() { + for(unsigned contacts : {0u,3u}) { + Fixture f; f.mesh.contacts=6; f.drain_usb=false; f.stream.write_capacity=0; + f.sendUsb({CMD_GET_CONTACTS}); f.step(); f.run(300); + assert(f.mesh._iter_started && f.mesh._iter_reply_route==&f.usb); + assert(f.usbFrames().empty()); // Old START/contacts are already queued. + assert(f.mesh.resetContactsForRecovery()); + assert(f.mesh._iter_started && f.mesh._iter_reply_route==&f.usb); + f.mesh.contacts=contacts; f.mesh.contact_base=64; + f.ble.input.push_back({CMD_GET_BATT_AND_STORAGE}); f.step(); f.run(100); + assert(count(f.ble.delivered,RESP_CODE_BATT_AND_STORAGE)==1); + assert(f.mesh._iter_started && f.usbFrames().empty()); + f.drain_usb=true; f.finish(); + assert(count(f.usbFrames(),RESP_CODE_CONTACTS_START)==2); + assertLatestSnapshot(f.usbFrames(),contacts,64); + assert(count(f.ble.delivered,RESP_CODE_CONTACTS_START)==0); + assert(count(f.ble.delivered,RESP_CODE_CONTACT)==0); + assert(count(f.ble.delivered,RESP_CODE_END_OF_CONTACTS)==0); + } + // No START has been admitted when recovery begins. It remains owned and is + // replaced by exactly one fresh marker/count once backpressure clears. + Fixture f; f.mesh.contacts=5; f.ble.capacity=0; f.startBle(); + assert(f.mesh._iter_start_pending); + assert(f.mesh.resetContactsForRecovery()); + assert(f.mesh._iter_started && f.mesh._iter_start_pending); + f.mesh.contacts=0; f.ble.capacity=4; f.finish(); + assert(count(f.ble.delivered,RESP_CODE_CONTACTS_START)==1); + assertLatestSnapshot(f.ble.delivered,0,64); +} + +static void recovered_stream_still_obeys_owner_disconnect_and_cancel() { + for(unsigned action=0;action!=3;++action) { + Fixture f; f.mesh.contacts=5; f.startBle(); f.ble.capacity=0; f.run(70); + assert(f.mesh.resetContactsForRecovery()); + assert(f.mesh._iter_started && f.mesh._iter_reply_route==&f.ble); + f.mesh.contacts=3; f.mesh.contact_base=64; f.ble.capacity=4; + if(action==0) { f.ble.connected=false; f.step(); } + else if(action==1) f.mesh.cancelSerialResponseStream(&f.ble); + else { f.ble.input.push_back({CMD_APP_START,0,0,0,0,0,0,0}); f.step(); } + assert(!f.mesh._iter_started && f.mesh._iter_reply_route==nullptr); + f.sendUsb({CMD_GET_BATT_AND_STORAGE}); f.step(); f.run(300); + assert(count(f.usbFrames(),RESP_CODE_BATT_AND_STORAGE)==1); + assert(count(f.usbFrames(),RESP_CODE_CONTACT)==0); + assert(count(f.ble.delivered,RESP_CODE_CONTACT)==0); + assert(count(f.ble.delivered,RESP_CODE_SELF_INFO)==unsigned(action==2)); + } +} + int main() { + recovery_keeps_ble_owner_and_restarts_under_backpressure(); + recovery_keeps_usb_owner_including_an_empty_reload(); + recovered_stream_still_obeys_owner_disconnect_and_cancel(); // Fifty-four BLE contacts need over three seconds, but each independent USB // request is handled immediately, even when it arrives on every loop pass. { @@ -238,12 +399,23 @@ int main() { assert(count(f.usbFrames(),RESP_CODE_END_OF_CONTACTS)==1); assert(count(f.ble.delivered,RESP_CODE_CONTACT)==0); } + puts("PASS: contact recovery restart, isolation and backpressure"); } ''' class ContactStreamConcurrencyTests(unittest.TestCase): def test_paced_stream_keeps_routes_without_blocking_other_clients(self): + for cache in (1, 0): + with self.subTest(MESH_CONTACT_CACHE=cache): + self.run_stream_test(cache) + + def test_stop_on_contact_recovery_negative_control(self): + for cache in (1, 0): + with self.subTest(MESH_CONTACT_CACHE=cache): + self.run_stream_test(cache, stop_on_recovery=True) + + def run_stream_test(self, cache, stop_on_recovery=False): compiler = os.environ.get('CXX') or shutil.which('g++') or shutil.which('clang++') if not compiler: self.skipTest('a host C++17 compiler is required') @@ -258,7 +430,16 @@ class ContactStreamConcurrencyTests(unittest.TestCase): methods = '\n'.join(extract_braced(mesh, sig) for sig in ( 'bool MyMesh::writeContactRespFrame(', 'void MyMesh::stopContactsIterator(', 'void MyMesh::cancelSerialResponseStream(', 'void MyMesh::checkSerialInterface()', + 'void MyMesh::onContactCacheFlushed()', 'bool MyMesh::resetContactsForRecovery()', )) + if stop_on_recovery: + # Restore the exact former cancellation while keeping table reset, + # lazy-state cleanup and transport code unchanged. + preparation = ('#if MESH_CONTACT_CACHE\n' + ' _iter_pending_contact = ContactInfo();\n' + '#endif\n _iter_contact_pending = false;') + self.assertEqual(methods.count(preparation), 1) + methods = methods.replace(preparation, 'stopContactsIterator();', 1) program = HARNESS.replace('@CONSTANTS@', constants).replace('@BUFFER@', extract_braced(native, 'class BufferStream') + ';') program = program.replace('@GET_CONTACTS@', contacts).replace('@APP_BEGIN@', app).replace('@METHODS@', methods) sanitizers = ['-fsanitize=address,undefined', '-fno-sanitize-recover=all', '-fno-pie', '-no-pie'] if sys.platform.startswith('linux') else [] @@ -267,13 +448,20 @@ class ContactStreamConcurrencyTests(unittest.TestCase): result = subprocess.run([ compiler, '-std=c++17', '-Wall', '-Wextra', '-Werror', '-Wno-unused-parameter', '-Wno-sign-compare', + f'-DMESH_CONTACT_CACHE={cache}', *sanitizers, f'-I{ROOT / "test/mocks"}', f'-I{ROOT / "src"}', '-x', 'c++', '-', str(ROOT / 'src/helpers/ArduinoSerialInterface.cpp'), '-o', str(binary), ], input=program, text=True, capture_output=True) self.assertEqual(result.returncode, 0, result.stderr) result = subprocess.run([str(binary)], text=True, capture_output=True) - self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + if stop_on_recovery: + self.assertNotEqual(result.returncode, 0) + self.assertIn('recovery_keeps_ble_owner_and_restarts_under_backpressure', result.stderr) + self.assertIn('f.mesh._iter_started', result.stderr) + else: + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertIn('PASS: contact recovery restart, isolation and backpressure', result.stdout) if __name__ == '__main__': diff --git a/test/test_companion_contact_stream_contract.py b/test/test_companion_contact_stream_contract.py index 3854f06f9..5a026fdc9 100644 --- a/test/test_companion_contact_stream_contract.py +++ b/test/test_companion_contact_stream_contract.py @@ -103,6 +103,26 @@ class ContactStreamContractTests(unittest.TestCase): ): self.assertLess(body.index(statement, restart), marker) + def test_contact_recovery_preserves_the_stream_owner_until_revision_restart(self): + impl = source("examples/companion_radio/MyMesh.cpp") + reset = function_body(impl, "bool MyMesh::resetContactsForRecovery()") + reset_at = reset.index("resetContacts()") + self.assertLess(reset.index("_iter_pending_contact = ContactInfo()"), reset_at) + self.assertIn("#if MESH_CONTACT_CACHE\n _iter_pending_contact = ContactInfo();\n#endif", + reset) + self.assertLess(reset.index("_iter_contact_pending = false"), reset_at) + for cancelled in ("stopContactsIterator()", "_iter_started =", + "_iter_reply_route =", "_iter_table_revision ="): + self.assertNotIn(cancelled, reset) + # Deliberate session/identity/factory resets still cancel their owner. + framed = function_body(impl, "void MyMesh::handleCmdFrame(") + start = framed.index("cmd_frame[0] == CMD_IMPORT_PRIVATE_KEY") + end = framed.index("cmd_frame[0] == CMD_SEND_RAW_DATA", start) + self.assertIn("stopContactsIterator()", framed[start:end]) + cancellation = function_body(impl, "void MyMesh::cancelSerialResponseStream(") + self.assertIn("route == NULL || _iter_reply_route == route", cancellation) + self.assertIn("stopContactsIterator()", cancellation) + def test_disconnected_flow_control_reports_required_frame_failure(self): impl = source("src/helpers/ArduinoSerialInterface.cpp") body = function_body(