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

@@ -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

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;

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)

test/test_companion_contact_stream_concurrency.py

@@ -21,8 +21,12 @@ HARNESS = r'''
 #include <cstring>
 #include <deque>
 #include <limits>
+#if MESH_CONTACT_CACHE
+#include <memory>
+#endif
 #include <vector>
 #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<unsigned> 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<class T> bool hasNext(T*,ContactInfo& c) {
     if(next==count) return false;
-    c=ContactInfo(); c.id.pub_key[0]=static_cast<uint8_t>(++next);
+    c=ContactInfo(); c.id.pub_key[0]=static_cast<uint8_t>(base + ++next);
+#if MESH_CONTACT_CACHE
+    c.path_owner=std::make_shared<unsigned>(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<int32_t>(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<std::vector<uint8_t>>& 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<MAX_PATH_SIZE;++i) assert((*frame)[36+i]==base+rows);
+    } else if((*frame)[0]==RESP_CODE_END_OF_CONTACTS) {
+      assert(++ends==1 && frame+1==frames.end());
+    } else assert(false);
+  }
+  assert(rows==contacts && ends==1);
+}
+
+static void recovery_keeps_ble_owner_and_restarts_under_backpressure() {
+  Fixture f; f.mesh.contacts=5; f.startBle();
+  f.ble.capacity=0; f.run(70);
+  assert(f.mesh._iter_contact_pending);
+#if MESH_CONTACT_CACHE
+  assert(f.mesh._iter_pending_contact.path_owner);
+  std::weak_ptr<unsigned> 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__':

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(