diff --git a/examples/companion_radio/DataStore.cpp b/examples/companion_radio/DataStore.cpp index 8e74b1118..85af1d074 100644 --- a/examples/companion_radio/DataStore.cpp +++ b/examples/companion_radio/DataStore.cpp @@ -1,6 +1,7 @@ #include #include #include +#include #include "DataStore.h" #include "PrefsStorageLayout.h" #include @@ -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(*_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 diff --git a/examples/companion_radio/DataStore.h b/examples/companion_radio/DataStore.h index cb8fadb6d..52a1cd57c 100644 --- a/examples/companion_radio/DataStore.h +++ b/examples/companion_radio/DataStore.h @@ -9,6 +9,9 @@ #include #endif #include +#if MESH_CONTACT_CACHE && defined(ESP32_PLATFORM) +namespace mesh { class ContactFileTransaction; } +#endif #if COMPANION_FEATURE_READER #include #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(); diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 40e497394..10716de54 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/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 diff --git a/src/helpers/ContactFileTransaction.h b/src/helpers/ContactFileTransaction.h index 4a8cfedec..0ccef0e90 100644 --- a/src/helpers/ContactFileTransaction.h +++ b/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 diff --git a/src/helpers/ContactPathCache.h b/src/helpers/ContactPathCache.h index 5949326d9..f9556a3fc 100644 --- a/src/helpers/ContactPathCache.h +++ b/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(); diff --git a/test/fixtures/contact_cache/test_contact_cache.cpp b/test/fixtures/contact_cache/test_contact_cache.cpp index f42f951e4..521085439 100644 --- a/test/fixtures/contact_cache/test_contact_cache.cpp +++ b/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 _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 rename_snapshots; + std::map 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::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(); diff --git a/test/test_contact_cache.py b/test/test_contact_cache.py index 35b99b512..ca4c1724d 100644 --- a/test/test_contact_cache.py +++ b/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"