Service cached contact persistence between companion loop passes

examples/companion_radio/DataStore.cpp

@@ -1,6 +1,7 @@
 #include <Arduino.h>
 #include <stdlib.h>
 #include <initializer_list>
+#include <new>
 #include "DataStore.h"
 #include "PrefsStorageLayout.h"
 #include <helpers/FileRead.h>
@@ -53,6 +54,12 @@ DataStore::DataStore(FILESYSTEM& fs, mesh::RTCClock& clock) : _fs(&fs), _fsExtra
 {
 }
 
+DataStore::~DataStore() {
+#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  cancelContactWrite();
+#endif
+}
+
 #if defined(EXTRAFS) || defined(QSPIFLASH)
 DataStore::DataStore(FILESYSTEM& fs, FILESYSTEM& fsExtra, mesh::RTCClock& clock) : _fs(&fs), _fsExtra(&fsExtra),
     _configuredFsExtra(&fsExtra), _clock(&clock),
@@ -137,6 +144,12 @@ static bool contactPathPresence(FILESYSTEM* fs, const char* path,
 #endif
 
 void DataStore::begin() {
+#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  // Reinitialization must not retain a streaming transaction or an inode
+  // opened through an earlier filesystem route. Dirty mutations stay pending.
+  cancelContactWrite();
+  _contact_path_reader.close();
+#endif
 #if defined(RP2040_PLATFORM)
   identity_store.begin();
 #endif
@@ -548,6 +561,10 @@ void DataStore::useVolatilePrimaryFS(FILESYSTEM& fs) {
 #endif
 
 void DataStore::disableSecondaryFS(bool authority_unknown) {
+#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  cancelContactWrite();
+  _contact_path_reader.close();
+#endif
   _fsExtra = nullptr;
 #if defined(NRF52_PLATFORM)
   if (authority_unknown) {
@@ -708,6 +725,7 @@ bool DataStore::removeFile(FILESYSTEM* fs, const char* filename) {
 
 bool DataStore::formatFileSystem() {
 #if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  cancelContactWrite();
   _contact_path_reader.close();
 #endif
 #if defined(NRF52_PLATFORM) || defined(STM32_PLATFORM)
@@ -779,6 +797,7 @@ bool DataStore::formatFileSystem() {
     _channel_recovery_source = nullptr;
 #if MESH_CONTACT_CACHE
     _cache_load_incomplete = false;
+    _contact_write_requested = false;
 #else
     _uncached_contact_load_incomplete = false;
 #endif
@@ -1900,6 +1919,9 @@ bool DataStore::writeContactPage(DataStoreHost* host, uint8_t page,
 #endif
 
 void DataStore::loadContacts(DataStoreHost* host) {
+#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  cancelContactWrite();
+#endif
 #if !defined(NRF52_PLATFORM) && !MESH_CONTACT_CACHE
   if (_uncached_contact_load_incomplete) return;
 #endif
@@ -2207,6 +2229,9 @@ void DataStore::loadContacts(DataStoreHost* host) {
 }
 
 bool DataStore::saveContacts(DataStoreHost* host, bool (*filter)(const ContactInfo& c)) {
+#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  cancelContactWrite();
+#endif
   if (hasIncompleteContactLoad()) return false;
 #if defined(NRF52_PLATFORM)
   bool success = true;
@@ -2218,6 +2243,9 @@ bool DataStore::saveContacts(DataStoreHost* host, bool (*filter)(const ContactIn
   }
   return flushContactWrites(host, filter) && success;
 #elif MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  // Explicit save/flush callers require a fully durable current table. An
+  // unfinished lazy transaction may contain an earlier revision: discard it
+  // before synchronously writing the current contacts through the same format.
   auto& paths = mesh::contactPathStorage();
   paths.beginCommit();
   for (uint32_t i = 0;; ++i) {
@@ -2227,12 +2255,14 @@ bool DataStore::saveContacts(DataStoreHost* host, bool (*filter)(const ContactIn
   }
   if (!paths.preserveSnapshots(0, 0x8000)) {
     paths.endCommit(false);
+    _contact_write_requested = true;
     return false;
   }
   // The cold-path reader reuses one SPIFFS File during this streaming write.
   // Close it before replacing the original name so later reads reopen the
   // committed file. No complete contact-table copy is needed in RAM.
-  mesh::ContactFileTransaction writer(_getContactsChannelsFS(), "/contacts3");
+  mesh::ContactFileTransaction writer(_getContactsChannelsFS(), "/contacts3",
+                                      companionPathPresence);
   bool success = writer;
   uint8_t record[mesh::storage::CONTACT_RECORD_SIZE];
   for (uint32_t i = 0; success; ++i) {
@@ -2254,6 +2284,7 @@ bool DataStore::saveContacts(DataStoreHost* host, bool (*filter)(const ContactIn
     }
   }
   paths.endCommit(success);
+  _contact_write_requested = !success;
   return success;
 #else
 #if defined(STM32_PLATFORM)
@@ -2309,6 +2340,10 @@ bool DataStore::markContactDirty(const ContactInfo& contact) {
   return _dirty_contact_pages.mark(slot / mesh::storage::CONTACTS_PER_PAGE);
 #else
   (void)contact;
+#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  ++_contact_write_revision;
+  _contact_write_requested = true;
+#endif
   return true;
 #endif
 }
@@ -2322,6 +2357,10 @@ bool DataStore::releaseContact(const ContactInfo& contact) {
   return _dirty_contact_pages.mark(slot / mesh::storage::CONTACTS_PER_PAGE);
 #else
   (void)contact;
+#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  ++_contact_write_revision;
+  _contact_write_requested = true;
+#endif
   return true;
 #endif
 }
@@ -2389,6 +2428,129 @@ bool DataStore::truncateLegacyContacts(uint16_t remaining_contacts) {
 }
 #endif
 
+#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+void DataStore::cancelContactWrite() {
+  if (_contact_write == nullptr) return;
+  delete _contact_write;
+  _contact_write = nullptr;
+  _contact_path_reader.close();
+  mesh::contactPathStorage().endCommit(false);
+  _contact_write_host = nullptr;
+  _contact_write_filter = nullptr;
+}
+
+bool DataStore::cancelCooperativeWrite() {
+  // Cache pressure inside serialization/snapshot preservation must keep the
+  // reentrancy veto. Only an app/radio path update between service passes can
+  // abandon the lazy job before the ordinary guarded eviction flush.
+  if (_contact_write == nullptr || _contact_write_servicing) return false;
+  cancelContactWrite();
+  return true;
+}
+
+bool DataStore::serviceCachedContactWrite(DataStoreHost* host,
+                                         bool (*filter)(const ContactInfo&)) {
+  struct ServiceGuard {
+    bool& active;
+    explicit ServiceGuard(bool& value) : active(value) { active = true; }
+    ~ServiceGuard() { active = false; }
+  } guard(_contact_write_servicing);
+  if (hasIncompleteContactLoad()) {
+    cancelContactWrite();
+    return false;
+  }
+  _contact_write_requested = true;
+  if (_contact_write != nullptr
+      && (_contact_write_active_revision != _contact_write_revision
+          || _contact_write_host != host || _contact_write_filter != filter)) {
+    // A mutation can move records or replace path handles. Never publish the
+    // indices of an obsolete streaming snapshot. Keep dirty state armed and
+    // start the current revision on a later loop pass.
+    cancelContactWrite();
+    return true;
+  }
+  auto& paths = mesh::contactPathStorage();
+  if (_contact_write == nullptr) {
+    paths.beginCommit();
+    for (uint32_t i = 0;; ++i) {
+      auto* contact = host->getContactForStore(i);
+      if (contact == nullptr) break;
+      if (!filter || filter(*contact)) paths.mark(contact->path_ref.handle());
+    }
+    if (!paths.preserveSnapshots(0, 0x8000)) {
+      paths.endCommit(false);
+      return false;
+    }
+    _contact_write = new (std::nothrow) mesh::ContactFileTransaction(
+        _getContactsChannelsFS(), "/contacts3", companionPathPresence);
+    if (_contact_write == nullptr || !static_cast<bool>(*_contact_write)) {
+      delete _contact_write;
+      _contact_write = nullptr;
+      paths.endCommit(false);
+      return false;
+    }
+    _contact_write_host = host;
+    _contact_write_filter = filter;
+    _contact_write_active_revision = _contact_write_revision;
+    _contact_write_index = 0;
+    _contact_write_verifying = false;
+    return true;
+  }
+  if (!_contact_write_verifying) {
+    ContactInfo* contact;
+    do {
+      contact = host->getContactForStore(_contact_write_index++);
+    } while (contact != nullptr && filter && !filter(*contact));
+    if (contact != nullptr) {
+      uint8_t record[mesh::storage::CONTACT_RECORD_SIZE];
+      if (serializeContactRecord(*contact, record)
+          && _contact_write->write(record, sizeof(record)) == sizeof(record)) {
+        return true;
+      }
+      cancelContactWrite();
+      return false;
+    }
+    _contact_path_reader.close();
+    _contact_write_verifying = true;
+    return true;
+  }
+  if (_contact_write->readyToPublish()
+      && !paths.preserveSnapshots(0, 0x8000)) {
+    cancelContactWrite();
+    return false;
+  }
+  if (_contact_write->readyToPublish()) {
+    // Cold app reads and final snapshot preservation can reopen the original
+    // inode during verification. Close it at the actual publication boundary.
+    _contact_path_reader.close();
+  }
+  const auto progress = _contact_write->serviceCommit();
+  if (progress == mesh::ContactFileTransaction::CommitProgress::Pending) {
+    return true;
+  }
+  if (progress == mesh::ContactFileTransaction::CommitProgress::Failed) {
+    cancelContactWrite();
+    return false;
+  }
+  // serviceCommit's final name replacement and this handle publication are
+  // one non-yielding section. Every later cold read sees the corresponding
+  // committed file and source indices, including retained ContactInfo copies.
+  uint16_t index = 0;
+  for (uint32_t i = 0;; ++i) {
+    auto* contact = host->getContactForStore(i);
+    if (contact == nullptr) break;
+    if (!filter || filter(*contact)) paths.publish(contact->path_ref.handle(), index++);
+  }
+  paths.endCommit(true);
+  delete _contact_write;
+  _contact_write = nullptr;
+  _contact_write_host = nullptr;
+  _contact_write_filter = nullptr;
+  _contact_write_requested = false;
+  return true;
+}
+#endif
+
 bool DataStore::serviceContactWrites(DataStoreHost* host,
                                      bool (*filter)(const ContactInfo& c)) {
 #if defined(NRF52_PLATFORM)
@@ -2442,6 +2604,8 @@ bool DataStore::serviceContactWrites(DataStoreHost* host,
   if (!writeContactPage(host, (uint8_t)page, filter)) return false;
   _dirty_contact_pages.clear((uint8_t)page);
   return true;
+#elif MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  return serviceCachedContactWrite(host, filter);
 #else
   return saveContacts(host, filter);
 #endif
@@ -2449,6 +2613,9 @@ bool DataStore::serviceContactWrites(DataStoreHost* host,
 
 bool DataStore::flushContactWrites(DataStoreHost* host,
                                    bool (*filter)(const ContactInfo& c)) {
+#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  cancelContactWrite();
+#endif
   if (hasIncompleteContactLoad()) return false;
 #if defined(NRF52_PLATFORM)
   while (hasPendingContactWrites()) {
@@ -2464,6 +2631,8 @@ bool DataStore::hasPendingContactWrites() const {
 #if defined(NRF52_PLATFORM)
   return !hasIncompleteContactLoad()
       && (!_dirty_contact_pages.empty() || _legacy_contacts_pending_cleanup);
+#elif MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  return _contact_write_requested || _contact_write != nullptr;
 #else
   return false;
 #endif
@@ -2737,6 +2906,9 @@ void DataStore::checkAdvBlobFile() {
 }
 
 bool DataStore::migrateToSecondaryFS() {
+#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  cancelContactWrite();
+#endif
   if (_fsExtra == nullptr) return false;
 
   // Implemented below through verified copy transactions. On nRF52, all

examples/companion_radio/DataStore.h

@@ -9,6 +9,9 @@
 #include <helpers/ContactSecretCache.h>
 #endif
 #include <helpers/CompanionReaderConfig.h>
+#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+namespace mesh { class ContactFileTransaction; }
+#endif
 #if COMPANION_FEATURE_READER
 #include <helpers/bible/Reader.h>
 #endif
@@ -56,9 +59,24 @@ class DataStore
   bool _cache_load_incomplete = false;
 #if defined(ESP32_PLATFORM)
   File _contact_path_reader;
+  mesh::ContactFileTransaction* _contact_write = nullptr;
+  DataStoreHost* _contact_write_host = nullptr;
+  bool (*_contact_write_filter)(const ContactInfo&) = nullptr;
+  uint32_t _contact_write_revision = 0;
+  uint32_t _contact_write_active_revision = 0;
+  uint32_t _contact_write_index = 0;
+  bool _contact_write_verifying = false;
+  bool _contact_write_requested = false;
+  bool _contact_write_servicing = false;
+  void cancelContactWrite();
+  bool serviceCachedContactWrite(DataStoreHost* host,
+                                 bool (*filter)(const ContactInfo&));
 #endif
   bool readStoredPath(uint16_t source, uint8_t path[64]) override;
   bool flushCachedPaths() override;
+#if defined(ESP32_PLATFORM)
+  bool cancelCooperativeWrite() override;
+#endif
 #if MESH_CONTACT_SECRET_FLASH_CACHE
   uint32_t _secret_retry_at = 0;
   bool readSavedSecret(const uint8_t peer[32], const uint8_t identity[32],
@@ -103,6 +121,7 @@ class DataStore
 
 public:
   DataStore(FILESYSTEM& fs, mesh::RTCClock& clock);
+  ~DataStore();
   DataStore(FILESYSTEM& fs, FILESYSTEM& fsExtra, mesh::RTCClock& clock);
   void begin();
   bool formatFileSystem();

examples/companion_radio/MyMesh.cpp

@@ -225,7 +225,13 @@ static constexpr uint8_t DEFAULT_FEM_RX_GAIN = 1;
 #define DIRECT_SEND_PERHOP_FACTOR       6.0f
 #define DIRECT_SEND_PERHOP_EXTRA_MILLIS 250
 #define LAZY_CONTACTS_WRITE_DELAY       5000
-#define CONTACT_PAGE_WRITE_GAP          100
+#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM)
+  // A cached ESP save streams one record/CRC chunk per pass. Keep the full
+  // mesh loop responsive between steps without adding seconds of page delay.
+  #define CONTACT_PAGE_WRITE_GAP          1
+#else
+  #define CONTACT_PAGE_WRITE_GAP          100
+#endif
 #define EXPECTED_ACK_RETRY_RECHECK_MILLIS 1000
 #define EXPECTED_ACK_HOST_GRACE_MILLIS 10000
 #define EXPECTED_ACK_HOST_RECHECK_MILLIS 10

src/helpers/ContactFileTransaction.h

@@ -18,10 +18,15 @@ private:
   char _temp[48];
   char _backup[48];
   File _file;
+  File _verify;
   size_t _size = 0;
   uint32_t _crc = 0xffffffff;
   bool _ok = false;
   bool _finished = false;
+  enum class CommitStage : uint8_t { Flush, Open, Verify, Publish };
+  CommitStage _commit_stage = CommitStage::Flush;
+  uint32_t _verify_crc = 0xffffffff;
+  size_t _verify_remaining = 0;
   PresenceProbe _presence;
   static bool probe(FILESYSTEM* fs, const char* path, bool& present,
                     PresenceProbe presence) {
@@ -30,6 +35,7 @@ private:
     return true;
   }
 public:
+  enum class CommitProgress : uint8_t { Pending, Succeeded, Failed };
   static bool recover(FILESYSTEM* fs, const char* target,
                       PresenceProbe presence = nullptr) {
     char backup[48];
@@ -56,33 +62,65 @@ public:
   }
   ~ContactFileTransaction() {
     if (_file) _file.close();
+    if (_verify) _verify.close();
     if (!_finished) _fs->remove(_temp);
   }
   operator bool() const { return _ok; }
+  bool readyToPublish() const {
+    return !_finished && _commit_stage == CommitStage::Publish;
+  }
   size_t write(const uint8_t* data, size_t len) {
-    if (!_ok) return 0;
+    if (!_ok || _finished || _commit_stage != CommitStage::Flush) return 0;
     const size_t wrote = _file.write(data, len);
     _ok = wrote == len;
     _size += wrote;
     _crc = storage::updateCRC32(_crc, data, wrote);
     return wrote;
   }
-  bool commit(bool valid = true) {
-    if (_finished) return false;
-    if (_file) { _file.flush(); _file.close(); }
+  // Each verification pass reads at most one small chunk. The caller can
+  // service its transports between passes without exposing the rename gap.
+  // A synchronous durability operation uses commit() to drain the same steps.
+  CommitProgress serviceCommit(bool valid = true) {
+    if (_finished) return CommitProgress::Failed;
     bool ok = _ok && valid;
-    File verify = _fs->open(_temp, "r");
-    ok = ok && verify && verify.size() == _size;
-    uint32_t crc = 0xffffffff;
-    size_t remaining = _size;
-    uint8_t buf[64];
-    while (ok && remaining) {
-      const size_t count = remaining < sizeof(buf) ? remaining : sizeof(buf);
-      ok = verify.read(buf, count) == count;
-      if (ok) { crc = storage::updateCRC32(crc, buf, count); remaining -= count; }
+    if (ok && _commit_stage == CommitStage::Flush) {
+      if (_file) { _file.flush(); _file.close(); }
+      _commit_stage = CommitStage::Open;
+      return CommitProgress::Pending;
     }
-    if (verify) verify.close();
-    ok = ok && crc == _crc;
+    if (ok && _commit_stage == CommitStage::Open) {
+      _verify = _fs->open(_temp, "r");
+      ok = _verify && _verify.size() == _size;
+      if (ok) {
+        _verify_remaining = _size;
+        _commit_stage = CommitStage::Verify;
+        return CommitProgress::Pending;
+      }
+    }
+    if (ok && _commit_stage == CommitStage::Verify) {
+      uint8_t buf[64];
+      const size_t count = _verify_remaining < sizeof(buf)
+          ? _verify_remaining : sizeof(buf);
+      if (count != 0) {
+        ok = _verify.read(buf, count) == count;
+        if (ok) {
+          _verify_crc = storage::updateCRC32(_verify_crc, buf, count);
+          _verify_remaining -= count;
+          return CommitProgress::Pending;
+        }
+      } else {
+        _verify.close();
+        ok = _verify_crc == _crc;
+        if (ok) {
+          _commit_stage = CommitStage::Publish;
+          return CommitProgress::Pending;
+        }
+      }
+    }
+    // Keep both renames and any rollback in this pass. Cold-path readers must
+    // not run between removing the target name and installing its replacement.
+    if (_file) _file.close();
+    if (_verify) _verify.close();
     bool backup_exists = false, target_exists = false;
     if (ok) ok = probe(_fs, _backup, backup_exists, _presence)
         && probe(_fs, _target, target_exists, _presence);
@@ -97,7 +135,13 @@ public:
     if (ok) _fs->remove(_backup);
     else _fs->remove(_temp);
     _finished = true;
-    return ok;
+    return ok ? CommitProgress::Succeeded : CommitProgress::Failed;
+  }
+  bool commit(bool valid = true) {
+    CommitProgress progress;
+    do { progress = serviceCommit(valid); }
+    while (progress == CommitProgress::Pending);
+    return progress == CommitProgress::Succeeded;
   }
 };
 } // namespace mesh

src/helpers/ContactPathCache.h

@@ -14,6 +14,9 @@ class ContactPathBackend {
 public:
   virtual bool readStoredPath(uint16_t source, uint8_t path[64]) = 0;
   virtual bool flushCachedPaths() = 0;
+  // A lazy streaming transaction may be abandoned before the existing
+  // synchronous eviction flush. Synchronous/paged commits keep their veto.
+  virtual bool cancelCooperativeWrite() { return false; }
 };
 
 class ContactPathStorage {
@@ -123,6 +126,9 @@ class ContactPathPool : public ContactPathStorage {
   }
   int makeRoom() {
     int index = availableEntry();
+    if (index < 0 && _backend && _committing && !_flushing) {
+      _backend->cancelCooperativeWrite();
+    }
     if (index < 0 && _backend && !_committing && !_flushing) {
       _flushing = true;
       _backend->flushCachedPaths();

test/fixtures/contact_cache/test_contact_cache.cpp

@@ -16,6 +16,8 @@ class File {
   std::string _path;
   size_t _position = 0;
   bool _write = false;
+  unsigned _read_generation = 0;
+  std::vector<uint8_t> _read_inode;
 public:
   File() = default;
   explicit File(FakeFilesystem& fs) : _fs(&fs) {}
@@ -41,9 +43,13 @@ public:
   std::string fail_read;
   unsigned fail_rename = 0, renames = 0, writes = 0, reads = 0, opens = 0;
   std::vector<Files> rename_snapshots;
+  std::map<std::string, unsigned> generations;
   bool exists(const char* path) const { return files.count(path) != 0; }
   bool mkdir(const char*) { return true; }
-  bool remove(const char* path) { return files.erase(path) != 0; }
+  bool remove(const char* path) {
+    ++generations[path];
+    return files.erase(path) != 0;
+  }
   File open(const char* path, const char* mode = "r", bool = false) {
     ++opens;
     if (*mode != 'r') {
@@ -61,6 +67,7 @@ public:
 #endif
     files[to] = files.at(from);
     files.erase(from);
+    ++generations[from]; ++generations[to];
     rename_snapshots.push_back(files);
     return true;
   }
@@ -81,7 +88,12 @@ int _getLfsUsedBlockCount(FakeFilesystem* fs) {
 }
 
 File::File(FakeFilesystem* fs, const char* path, bool write)
-    : _fs(fs), _path(path), _write(write) {}
+    : _fs(fs), _path(path), _write(write) {
+  if (!write) {
+    _read_generation = fs->generations[path];
+    _read_inode = fs->files.at(path);
+  }
+}
 bool File::open(const char* path, uint8_t mode) {
   if (!_fs) return false;
   *this = _fs->open(path, mode == FILE_O_WRITE ? "w" : "r");
@@ -90,7 +102,10 @@ bool File::open(const char* path, uint8_t mode) {
 size_t File::read(uint8_t* bytes, size_t length) {
   if (!_fs || _write || _fs->fail_read == _path) return 0;
   ++_fs->reads;
-  auto& data = _fs->files[_path];
+  // An open file keeps its old inode across a rename/unlink. A later File
+  // opened on the same name observes the replacement, as SPIFFS/POSIX do.
+  auto& data = _fs->generations[_path] == _read_generation
+      ? _fs->files.at(_path) : _read_inode;
   const size_t count = std::min(length, data.size() - std::min(data.size(), _position));
   if (count) memcpy(bytes, data.data() + _position, count);
   _position += count;
@@ -106,7 +121,11 @@ size_t File::write(const uint8_t* bytes, size_t length) {
   _position += count;
   return count;
 }
-size_t File::size() const { return _fs ? _fs->files.at(_path).size() : 0; }
+size_t File::size() const {
+  if (!_fs) return 0;
+  return !_write && _fs->generations[_path] != _read_generation
+      ? _read_inode.size() : _fs->files.at(_path).size();
+}
 bool File::seek(size_t pos) {
   if (!_fs || pos > size()) return false;
   _position = pos;
@@ -160,7 +179,9 @@ class DataStore : public mesh::ContactPathBackend, public mesh::ContactSecretBac
   DataStoreHost* _cache_host;
   bool _cache_load_incomplete = false;
 #if defined(ESP32_PLATFORM)
-  File _contact_path_reader;
+#include "contact_write_state_under_test.h"
+  FakeFilesystem* _fs = &SPIFFS;
+  FakeFilesystem* _fsExtra = nullptr;
 #endif
   uint32_t _secret_retry_at = 0;
 #if defined(NRF52_PLATFORM)
@@ -178,7 +199,14 @@ public:
   explicit DataStore(Host& host) : _cache_host(&host) {
     mesh::contactPathStorage().attach(this);
   }
-  FakeFilesystem* _getContactsChannelsFS() const { return &SPIFFS; }
+  ~DataStore();
+  FakeFilesystem* _getContactsChannelsFS() const {
+#if defined(ESP32_PLATFORM)
+    return _fsExtra != nullptr ? _fsExtra : _fs;
+#else
+    return &SPIFFS;
+#endif
+  }
   File openRead(FakeFilesystem* fs, const char* path) { return fs->open(path); }
   bool hasIncompleteContactLoad() const {
     return _cache_load_incomplete
@@ -192,9 +220,21 @@ public:
   bool flushContactWrites(DataStoreHost*, bool (*filter)(const ContactInfo&));
   bool readStoredPath(uint16_t, uint8_t[64]) override;
   bool flushCachedPaths() override;
+#if defined(ESP32_PLATFORM)
+  void begin();
+  void disableSecondaryFS(bool);
+  void useSecondary(FakeFilesystem* fs) { _fsExtra = fs; }
+  bool cancelCooperativeWrite() override;
+#endif
   uint16_t secretSlot(const uint8_t[32]) const;
   bool readSavedSecret(const uint8_t[32], const uint8_t[32], uint8_t[32]) override;
   bool saveSecret(const uint8_t[32], const uint8_t[32], const uint8_t[32]) override;
+#if defined(ESP32_PLATFORM)
+  bool markContactDirty(const ContactInfo&);
+  bool releaseContact(const ContactInfo&);
+  bool hasPendingContactWrites() const;
+  bool serviceContactWrites(DataStoreHost*, bool (*filter)(const ContactInfo&));
+#endif
 #if defined(NRF52_PLATFORM)
   bool loadContactPages(DataStoreHost*, uint16_t, uint32_t);
   bool writeContactPage(DataStoreHost*, uint8_t, bool (*filter)(const ContactInfo&));
@@ -505,6 +545,180 @@ static void contact_reload_preserves_disk_on_cache_exhaustion() {
 }
 
 #if defined(ESP32_PLATFORM)
+static unsigned drain_cooperative_save(Fixture& f) {
+  unsigned serviced_commands = 0;
+  while (f.store.hasPendingContactWrites()) {
+    const unsigned before_writes = SPIFFS.writes;
+    const unsigned before_reads = SPIFFS.reads;
+    assert(f.store.serviceContactWrites(&f.host, cachedContactFilter));
+    // Returning from every bounded store pass lets the production mesh loop
+    // poll transports. No pass serializes the whole table or verifies it all.
+    assert(SPIFFS.writes - before_writes <= 1);
+    assert(SPIFFS.reads - before_reads <= 16); // retained path snapshots
+    ++serviced_commands;
+    assert(serviced_commands < 3000);
+    assert(SPIFFS.exists("/contacts3")); // no observable rename gap
+    for (unsigned i = 0; i < f.host.contacts.size(); i += 47) {
+      Path path;
+      assert(f.host.contacts[i].copyPathTo(path.data()));
+    }
+  }
+  return serviced_commands;
+}
+
+static void cooperative_save_services_commands_and_preserves_snapshots() {
+  Fixture f;
+  const ContactInfo snapshot = f.host.contacts[349];
+  const auto old = SPIFFS.files.at("/contacts3");
+  assert(f.host.contacts[349].setRawPath(route(980).data()));
+  assert(f.store.markContactDirty(f.host.contacts[349]));
+  assert(f.store.hasPendingContactWrites());
+  assert(f.store.serviceContactWrites(&f.host, cachedContactFilter));
+  assert(f.store.hasPendingContactWrites() && SPIFFS.files.at("/contacts3") == old);
+  const unsigned polled = drain_cooperative_save(f);
+  assert(polled > 350); // record writes and CRC readback both yield
+  assert(SPIFFS.files.at("/contacts3") != old);
+  f.check(349, route(980));
+  Path path;
+  assert(snapshot.copyPathTo(path.data()) && path == route(349));
+  for (const auto& files : SPIFFS.rename_snapshots) {
+    FakeFilesystem rebooted;
+    rebooted.files = files;
+    assert(mesh::ContactFileTransaction::recover(&rebooted, "/contacts3"));
+    assert(rebooted.files.at("/contacts3") == old
+        || rebooted.files.at("/contacts3") == SPIFFS.files.at("/contacts3"));
+  }
+}
+
+static void cooperative_mutations_never_publish_obsolete_indices() {
+  for (unsigned mutation_step : {4u, 46u, 110u, 140u}) {
+    Fixture f(40);
+    const auto old = SPIFFS.files.at("/contacts3");
+    assert(f.store.markContactDirty(f.host.contacts[0]));
+    for (unsigned step = 0; step < mutation_step; ++step) {
+      assert(f.store.serviceContactWrites(&f.host, cachedContactFilter));
+      assert(SPIFFS.files.at("/contacts3") == old);
+    }
+    const ContactInfo snapshot = f.host.contacts[3]; // created during save
+    assert(f.store.releaseContact(f.host.contacts[3]));
+    f.host.contacts.erase(f.host.contacts.begin() + 3); // shifts all later indices
+    assert(f.host.contacts[1].setRawPath(route(991).data()));
+    snprintf(f.host.contacts[1].name, sizeof(f.host.contacts[1].name), "new revision");
+    assert(f.store.markContactDirty(f.host.contacts[1]));
+    assert(f.store.serviceContactWrites(&f.host, cachedContactFilter));
+    assert(f.store.hasPendingContactWrites()); // abort is not durable success
+    assert(SPIFFS.files.at("/contacts3") == old);
+    drain_cooperative_save(f);
+    const auto& bytes = SPIFFS.files.at("/contacts3");
+    assert(bytes.size() == 39 * 152);
+    for (unsigned i = 0; i < 39; ++i) {
+      assert(memcmp(bytes.data() + i * 152, f.host.contacts[i].id.pub_key, 32) == 0);
+      Path path;
+      assert(f.host.contacts[i].copyPathTo(path.data()));
+      const Path expected = route(i == 1 ? 991 : i >= 3 ? i + 1 : i);
+      assert(path == expected);
+    }
+    Path path;
+    assert(snapshot.copyPathTo(path.data()) && path == route(3));
+  }
+}
+
+static void cooperative_failures_keep_the_old_file_and_pending_mutations() {
+  for (unsigned fault = 0; fault < 4; ++fault) {
+    Fixture f(30);
+    const auto old = SPIFFS.files.at("/contacts3");
+    assert(f.host.contacts[0].setRawPath(route(993).data()));
+    assert(f.store.markContactDirty(f.host.contacts[0]));
+    if (fault == 0) SPIFFS.max_write = 17;
+    if (fault == 1) SPIFFS.fail_read = "/contacts3.tmp";
+    if (fault == 2) SPIFFS.fail_rename = SPIFFS.renames + 2;
+    bool failed = false;
+    for (unsigned step = 0; step < 1000 && !failed; ++step) {
+      if (fault == 3 && step == 35) SPIFFS.files.at("/contacts3.tmp")[100] ^= 1;
+      failed = !f.store.serviceContactWrites(&f.host, cachedContactFilter);
+    }
+    assert(failed && f.store.hasPendingContactWrites());
+    assert(SPIFFS.files.at("/contacts3") == old);
+    assert(!SPIFFS.exists("/contacts3.tmp"));
+    f.check(0, route(993));
+    SPIFFS.max_write = std::numeric_limits<size_t>::max();
+    SPIFFS.fail_read.clear(); SPIFFS.fail_rename = 0;
+    drain_cooperative_save(f);
+    assert(SPIFFS.files.at("/contacts3") != old);
+    f.check(0, route(993));
+  }
+}
+
+static void synchronous_flush_cancels_pending_jobs_and_stays_durable() {
+  Fixture f(40);
+  const ContactInfo snapshot = f.host.contacts[0];
+  assert(f.host.contacts[0].setRawPath(route(994).data()));
+  assert(f.store.markContactDirty(f.host.contacts[0]));
+  for (unsigned i = 0; i < 8; ++i)
+    assert(f.store.serviceContactWrites(&f.host, cachedContactFilter));
+  // Update, reorder, then force explicit durability instead of draining a
+  // snapshot from an earlier revision. The rename callback sees a full file.
+  std::swap(f.host.contacts[0], f.host.contacts[3]);
+  assert(f.store.markContactDirty(f.host.contacts[0]));
+  assert(f.store.flushContactWrites(&f.host, cachedContactFilter));
+  assert(!f.store.hasPendingContactWrites() && !SPIFFS.exists("/contacts3.tmp"));
+  assert(memcmp(SPIFFS.files.at("/contacts3").data(), f.host.contacts[0].id.pub_key, 32) == 0);
+  f.check(3, route(994));
+  Path path;
+  assert(snapshot.copyPathTo(path.data()) && path == route(0));
+}
+
+static void interleaved_path_pressure_preserves_guarded_eviction() {
+  Fixture f(40);
+  for (unsigned i = 0; i < 26; ++i) {
+    // Keep a real lazy job open while more than all16 resident path slots are
+    // replaced by app updates. makeRoom must abandon only that cooperative
+    // job before the pre-existing synchronous eviction flush obtains room.
+    if (!f.store.hasPendingContactWrites())
+      assert(f.store.markContactDirty(f.host.contacts[0]));
+    assert(f.store.serviceContactWrites(&f.host, cachedContactFilter));
+    assert(f.host.contacts[i].setRawPath(route(1200 + i).data()));
+    assert(f.store.markContactDirty(f.host.contacts[i]));
+  }
+  assert(f.host.cache_flushes > 0); // actual cache exhaustion, no mocked bypass
+  drain_cooperative_save(f);
+  for (unsigned i = 0; i < 40; ++i)
+    f.check(i, route(i < 26 ? 1200 + i : i));
+}
+
+static void lifecycle_cancels_jobs_and_closes_prior_filesystem_readers() {
+  {
+    Fixture f(30);
+    assert(f.store.markContactDirty(f.host.contacts[0]));
+    assert(f.store.serviceContactWrites(&f.host, cachedContactFilter));
+    f.check(29, route(29)); // opens the original cold-path reader
+    f.store.begin();
+    assert(f.store.hasPendingContactWrites());
+    assert(!SPIFFS.exists("/contacts3.tmp"));
+    const unsigned opens = SPIFFS.opens;
+    f.check(28, route(28));
+    assert(SPIFFS.opens == opens + 1); // reinitialization reopened its reader
+    drain_cooperative_save(f);
+  }
+  {
+    Fixture f(30);
+    FakeFilesystem alternate;
+    alternate.files = SPIFFS.files;
+    f.store.useSecondary(&alternate);
+    assert(f.store.markContactDirty(f.host.contacts[0]));
+    assert(f.store.serviceContactWrites(&f.host, cachedContactFilter));
+    f.check(29, route(29));
+    const unsigned opens = SPIFFS.opens;
+    f.store.disableSecondaryFS(false);
+    assert(f.store.hasPendingContactWrites());
+    assert(!alternate.exists("/contacts3.tmp"));
+    f.check(28, route(28));
+    assert(SPIFFS.opens == opens + 1); // reads follow the primary route
+    drain_cooperative_save(f);
+    // No lazy job remains tied to alternate's shorter lifetime.
+  }
+}
+
 static void partial_loads_cannot_replace_the_complete_file() {
   for (bool capacity_failure : {false, true}) {
     Fixture f(40);
@@ -667,6 +881,12 @@ static void contact_tx_policy_round_trips_without_growing_records() {
 int main() {
   contact_tx_policy_round_trips_without_growing_records();
 #if defined(ESP32_PLATFORM)
+  cooperative_save_services_commands_and_preserves_snapshots();
+  cooperative_mutations_never_publish_obsolete_indices();
+  cooperative_failures_keep_the_old_file_and_pending_mutations();
+  synchronous_flush_cancels_pending_jobs_and_stays_durable();
+  interleaved_path_pressure_preserves_guarded_eviction();
+  lifecycle_cancels_jobs_and_closes_prior_filesystem_readers();
   routes_survive_eviction_and_full_sync();
   snapshot_rollback_and_dirty_eviction();
   storage_failures_preserve_routes();

test/test_contact_cache.py

@@ -63,6 +63,24 @@ class ContactCacheTest(unittest.TestCase):
             helper_start = store.index("namespace {\nbool cachedContactFilter")
             helper_end = store.index("bool DataStore::readStoredPath(", helper_start)
             implementation = store[helper_start:helper_end]
+            header = (ROOT / "examples/companion_radio/DataStore.h").read_text()
+            state_start = header.index("  File _contact_path_reader;")
+            state_end = header.index("\n#endif", state_start)
+            (temp / "contact_write_state_under_test.h").write_text(
+                header[state_start:state_end])
+            functions.append("DataStore::~DataStore()")
+            if platform == "ESP32_PLATFORM":
+                functions.extend((
+                    "void DataStore::cancelContactWrite(",
+                    "bool DataStore::cancelCooperativeWrite(",
+                    "bool DataStore::serviceCachedContactWrite(",
+                    "bool DataStore::markContactDirty(",
+                    "bool DataStore::releaseContact(",
+                    "bool DataStore::serviceContactWrites(",
+                    "bool DataStore::hasPendingContactWrites() const",
+                    "void DataStore::begin()",
+                    "void DataStore::disableSecondaryFS(",
+                ))
             if platform == "NRF52_PLATFORM":
                 implementation += method(store, "static void makeContactPagePath(") + "\n"
                 implementation += method(store, "static void discardInvalidContactPage(") + "\n"