diff --git a/examples/companion_radio/DataStore.cpp b/examples/companion_radio/DataStore.cpp index 8e43bdb2c..d2863c41c 100644 --- a/examples/companion_radio/DataStore.cpp +++ b/examples/companion_radio/DataStore.cpp @@ -6,6 +6,7 @@ #include "PrefsStorageLayout.h" #include #include +#include #if defined(ESP32_PLATFORM) || defined(RP2040_PLATFORM) #include #endif @@ -145,6 +146,7 @@ static bool contactPathPresence(FILESYSTEM* fs, const char* path, void DataStore::begin() { #if defined(ESP32_PLATFORM) + _advert_write.clear(); // Reinitialization must not retain a streaming transaction or an inode // opened through an earlier filesystem route. Dirty mutations stay pending. cancelContactWrite(); @@ -729,6 +731,8 @@ bool DataStore::removeFile(FILESYSTEM* fs, const char* filename) { bool DataStore::formatFileSystem() { #if defined(ESP32_PLATFORM) + // A failed format keeps the latest accepted RAM packets available to retry. + _advert_write.pause(); cancelContactWrite(); #if MESH_CONTACT_CACHE _contact_path_reader.close(); @@ -796,6 +800,7 @@ bool DataStore::formatFileSystem() { bool fs_success = ((fs::SPIFFSFS *)_fs)->format(); esp_err_t nvs_err = nvs_flash_erase(); // no need to reinit, will be done by reboot if (fs_success && nvs_err == ESP_OK) { + _advert_write.clear(); _identity_creation_blocked = false; _prefs_load_incomplete = false; _channel_load_incomplete = false; @@ -3761,9 +3766,34 @@ inline void makeBlobPath(const uint8_t key[], int key_len, char* path, size_t pa uint8_t DataStore::getBlobByKey(const uint8_t key[], int key_len, uint8_t dest_buf[]) { char path[64]; +#if defined(ESP32_PLATFORM) + if (key_len <= 0) return 0; +#endif makeBlobPath(key, key_len, path, sizeof(path)); - if (_fs->exists(path)) { +#if defined(ESP32_PLATFORM) + const int prefix_len = key_len > 8 ? 8 : key_len; + for (const auto& slot : _advert_write.slots) { + if (slot.len && slot.key_len == prefix_len + && memcmp(slot.key, key, prefix_len) == 0) { + memcpy(dest_buf, slot.bytes, slot.len); + return slot.len; + } + } + // SPIFFS cannot rename over an existing file. A reset between the staged + // target->backup and temp->target renames leaves the prior valid cache here. + _advert_write.synchronous_io = true; + bool present = false; + if (!companionPathPresence(_fs, path, present)) return 0; + if (!present) { + const size_t used = strlen(path); + strcpy(path + used, ".bak"); + if (!companionPathPresence(_fs, path, present)) return 0; + } +#else + const bool present = _fs->exists(path); +#endif + if (present) { File f = openRead(_fs, path); if (f) { int len = f.read(dest_buf, 255); // currently MAX 255 byte blob len supported!! @@ -3777,12 +3807,21 @@ uint8_t DataStore::getBlobByKey(const uint8_t key[], int key_len, uint8_t dest_b bool DataStore::putBlobByKey(const uint8_t key[], int key_len, const uint8_t src_buf[], uint8_t len) { char path[64]; makeBlobPath(key, key_len, path, sizeof(path)); +#if defined(ESP32_PLATFORM) + _advert_write.synchronous_io = true; + invalidateAdvertWrite(key, key_len); +#endif File f = openWrite(_fs, path); if (f) { int n = f.write(src_buf, len); f.close(); - if (n == len) return true; // success! + if (n == len) { +#if defined(ESP32_PLATFORM) + retireAdvertWrite(key, key_len); +#endif + return true; + } _fs->remove(path); // blob was only partially written! } @@ -3794,12 +3833,23 @@ bool DataStore::deleteBlobByKey(const uint8_t key[], int key_len) { makeBlobPath(key, key_len, path, sizeof(path)); #if defined(ESP32_PLATFORM) + _advert_write.synchronous_io = true; + invalidateAdvertWrite(key, key_len); // VFS remove() logs an error for normal absence, contaminating the binary // USB stream. Distinguish absence from metadata failure before calling it; // a real deletion failure must preserve the live contact for rollback. - bool present = false; - if (!companionPathPresence(_fs, path, present)) return false; - return !present || _fs->remove(path); + char backup[68]; + snprintf(backup, sizeof(backup), "%s.bak", path); + bool present = false, backup_present = false; + // Probe both before mutating either; removing the backup first preserves + // the authoritative target if a later deletion fails and the contact rolls + // back. A successful deletion invalidates publication before staged cleanup. + if (!companionPathPresence(_fs, path, present) + || !companionPathPresence(_fs, backup, backup_present)) return false; + if (backup_present && !_fs->remove(backup)) return false; + if (present && !_fs->remove(path)) return false; + retireAdvertWrite(key, key_len); + return true; #else _fs->remove(path); @@ -3808,6 +3858,273 @@ bool DataStore::deleteBlobByKey(const uint8_t key[], int key_len) { } #endif +#if defined(ESP32_PLATFORM) +bool DataStore::queueAdvertByKey(const uint8_t key[], int key_len, + const uint8_t src_buf[], uint8_t len) { + if (key_len <= 0 || len == 0) return false; + const uint8_t prefix_len = key_len > 8 ? 8 : key_len; + AdvertWriteState::Slot* chosen = nullptr; + for (auto& slot : _advert_write.slots) { + if (slot.len && slot.key_len == prefix_len + && memcmp(slot.key, key, prefix_len) == 0) { + chosen = &slot; + break; + } + if (!slot.len && chosen == nullptr) chosen = &slot; + } + // Never force a filesystem flush or allocate another packet on overflow. + if (chosen == nullptr) return false; + memcpy(chosen->key, key, prefix_len); + memcpy(chosen->bytes, src_buf, len); + chosen->key_len = prefix_len; + chosen->len = len; + chosen->revision = ++_advert_write.revision; + return true; +} + +void DataStore::retireAdvertWrite(const uint8_t key[], int key_len) { + if (key_len <= 0) return; + const int prefix_len = key_len > 8 ? 8 : key_len; + for (auto& slot : _advert_write.slots) { + if (slot.len && slot.key_len == prefix_len + && memcmp(slot.key, key, prefix_len) == 0) { + slot.len = 0; + slot.revision = ++_advert_write.revision; + } + } +} + +void DataStore::invalidateAdvertWrite(const uint8_t key[], int key_len) { + if (key_len <= 0) return; + const int prefix_len = key_len > 8 ? 8 : key_len; + auto& job = _advert_write; + for (auto& slot : job.slots) { + if (slot.len && slot.key_len == prefix_len + && memcmp(slot.key, key, prefix_len) == 0) { + slot.revision = ++job.revision; + } + } + if (job.stage != AdvertWriteState::Stage::Idle + && job.active_key_len == prefix_len + && memcmp(job.active_key, key, prefix_len) == 0) { + // A synchronous API may have changed target/backup even when it returned + // failure. Re-probe the independent active key before recovery/publication. + job.cancelling = true; + job.stage = AdvertWriteState::Stage::CancelClose; + job.retry_at = 0; + } +} + +bool DataStore::consumeSynchronousAdvertIO() { + const bool result = _advert_write.synchronous_io; + _advert_write.synchronous_io = false; + return result; +} + +bool DataStore::hasPendingAdvertWrites() const { + if (_advert_write.stage != AdvertWriteState::Stage::Idle) return true; + for (const auto& slot : _advert_write.slots) if (slot.len) return true; + return false; +} + +bool DataStore::isAdvertWriteDue(uint32_t now) const { + return hasPendingAdvertWrites() && (_advert_write.retry_at == 0 + || static_cast(now - _advert_write.retry_at) >= 0); +} + +bool DataStore::serviceAdvertWrites(uint32_t now) { + auto& job = _advert_write; + using Stage = AdvertWriteState::Stage; + static const char* const temp = "/advert.tmp"; + if (!isAdvertWriteDue(now)) return true; + job.retry_at = 0; + if (job.stage == Stage::Idle) { + for (uint8_t n = 0; n < AdvertWriteState::CAPACITY; ++n) { + const uint8_t i = (job.next + n) % AdvertWriteState::CAPACITY; + if (!job.slots[i].len) continue; + job.active = i; + job.next = (i + 1) % AdvertWriteState::CAPACITY; + job.active_revision = job.slots[i].revision; + job.active_key_len = job.slots[i].key_len; + memcpy(job.active_key, job.slots[i].key, job.active_key_len); + makeBlobPath(job.slots[i].key, job.slots[i].key_len, + job.target, sizeof(job.target)); + snprintf(job.backup, sizeof(job.backup), "%s.bak", job.target); + job.offset = 0; + job.cancelling = false; + job.stage = Stage::TargetProbe; + break; + } + if (job.stage == Stage::Idle) return true; + } + auto& slot = job.slots[job.active]; + if (!job.cancelling && (!slot.len || slot.revision != job.active_revision)) { + job.cancelling = true; + job.stage = Stage::CancelClose; + } + const auto retry = [&]() { + const uint32_t delay = mesh::recordLazyPersistenceSaveFailure( + job.failures, 1000, mesh::LAZY_PERSISTENCE_MAX_RETRY_DELAY_MILLIS); + job.retry_at = now + delay; + if (job.retry_at == 0) job.retry_at = 1; + return false; + }; + const auto fail = [&]() { + job.cancelling = true; + job.stage = Stage::CancelClose; + return retry(); + }; + const auto prepared = [&]() { + if (job.cancelling) { + job.stage = Stage::Idle; + job.active = AdvertWriteState::CAPACITY; + } else job.stage = Stage::OpenWrite; + }; + // Exactly one filesystem API operation in each arm. In particular, open, + // fclose (which may flush stdio), crypto in BaseChatMesh, and contact-file + // work must not accumulate in the same serviced loop pass. SPIFFS may still + // spend an unbounded amount of wall time inside one individual operation. + switch (job.stage) { + case Stage::TargetProbe: + if (!companionPathPresence(_fs, job.target, job.target_present)) return fail(); + job.stage = Stage::BackupProbe; + break; + case Stage::BackupProbe: + if (!companionPathPresence(_fs, job.backup, job.backup_present)) return fail(); + job.stage = job.backup_present ? (job.target_present ? Stage::RemoveBackup + : Stage::RestoreBackup) + : Stage::TempProbe; + break; + case Stage::RestoreBackup: + if (!_fs->rename(job.backup, job.target)) return fail(); + job.stage = Stage::TempProbe; + break; + case Stage::RemoveBackup: + if (!_fs->remove(job.backup)) return fail(); + job.stage = Stage::TempProbe; + break; + case Stage::TempProbe: + if (!companionPathPresence(_fs, temp, job.temp_present)) return fail(); + if (job.temp_present) job.stage = Stage::RemoveTemp; + else prepared(); + break; + case Stage::RemoveTemp: + if (!_fs->remove(temp)) return fail(); + prepared(); + break; + case Stage::OpenWrite: + job.file = openWrite(_fs, temp); + if (!job.file) return fail(); + job.stage = Stage::ConfigureWrite; + break; + case Stage::ConfigureWrite: + if (!job.file.setBufferSize(sizeof(job.scratch))) return fail(); + job.stage = Stage::Write; + break; + case Stage::Write: { + const uint8_t count = static_cast(slot.len - job.offset) > sizeof(job.scratch) + ? sizeof(job.scratch) : slot.len - job.offset; + if (job.file.write(slot.bytes + job.offset, count) != count) return fail(); + job.offset += count; + if (job.offset == slot.len) job.stage = Stage::CloseWrite; + break; + } + case Stage::CloseWrite: + job.file.close(); + job.stage = Stage::OpenVerify; + break; + case Stage::OpenVerify: + job.file = openRead(_fs, temp); + if (!job.file) return fail(); + job.stage = Stage::ConfigureVerify; + break; + case Stage::ConfigureVerify: + if (!job.file.setBufferSize(sizeof(job.scratch))) return fail(); + job.stage = Stage::VerifySize; + break; + case Stage::VerifySize: + if (job.file.size() != slot.len) return fail(); + job.offset = 0; + job.stage = Stage::Verify; + break; + case Stage::Verify: { + const uint8_t count = static_cast(slot.len - job.offset) > sizeof(job.scratch) + ? sizeof(job.scratch) : slot.len - job.offset; + if (job.file.read(job.scratch, count) != count + || memcmp(job.scratch, slot.bytes + job.offset, count) != 0) return fail(); + job.offset += count; + if (job.offset == slot.len) job.stage = Stage::CloseVerify; + break; + } + case Stage::CloseVerify: + job.file.close(); + job.stage = Stage::PublishTargetProbe; + break; + case Stage::PublishTargetProbe: + if (!companionPathPresence(_fs, job.target, job.target_present)) return fail(); + job.stage = Stage::PublishBackupProbe; + break; + case Stage::PublishBackupProbe: + if (!companionPathPresence(_fs, job.backup, job.backup_present)) return fail(); + job.stage = job.target_present ? (job.backup_present ? Stage::PublishRemoveBackup + : Stage::PublishBackup) + : Stage::Publish; + break; + case Stage::PublishRemoveBackup: + if (!_fs->remove(job.backup)) return fail(); + job.stage = Stage::PublishBackup; + break; + case Stage::PublishBackup: + if (!_fs->rename(job.target, job.backup)) return fail(); + job.backup_present = true; + job.stage = Stage::Publish; + break; + case Stage::Publish: + // SPIFFS rejects rename over an existing destination. The two renames + // have an explicit gap; RAM serves the new packet and disk readers after + // a reset use the old backup. No unverified temp is ever a visible blob. + if (!_fs->rename(temp, job.target)) return fail(); + job.stage = Stage::PublishCleanup; + break; + case Stage::PublishCleanup: + if (job.backup_present && !_fs->remove(job.backup)) return retry(); + slot.len = 0; + job.failures = 0; + job.stage = Stage::Idle; + job.active = AdvertWriteState::CAPACITY; + break; + case Stage::CancelClose: + job.file.close(); + job.stage = Stage::TargetProbe; + break; + case Stage::Idle: + break; + } + return true; +} + +bool DataStore::flushAdvertWrites(uint32_t now) { + // Explicit reboot/shutdown/import paths require durability and may block. + // On failure keep accepted packets in RAM, but never leave a file open. + _advert_write.synchronous_io = hasPendingAdvertWrites() + || _advert_write.synchronous_io; + for (unsigned pass = 0; hasPendingAdvertWrites() && pass < 512; ++pass) { + _advert_write.retry_at = 0; + if (!serviceAdvertWrites(now)) { + const uint32_t retry_at = _advert_write.retry_at; + _advert_write.pause(); + _advert_write.retry_at = retry_at; + return false; + } + } + if (hasPendingAdvertWrites()) { + _advert_write.pause(); + return false; + } + return true; +} +#endif + #if MESH_CONTACT_CACHE namespace { bool cachedContactFilter(const ContactInfo& c) { return c.type != ADV_TYPE_NONE; } diff --git a/examples/companion_radio/DataStore.h b/examples/companion_radio/DataStore.h index 422265160..30ef3cc70 100644 --- a/examples/companion_radio/DataStore.h +++ b/examples/companion_radio/DataStore.h @@ -60,6 +60,59 @@ class DataStore bool serviceContactWrite(DataStoreHost* host, bool (*filter)(const ContactInfo&)); #endif +#if defined(ESP32_PLATFORM) + // Advert packets are reconstructable cache data. Keep accepted updates in + // RAM while their filesystem operations run on separate loop passes. + struct AdvertWriteState { + static constexpr uint8_t CAPACITY = 4; + struct Slot { + uint8_t key[8] = {}; + uint8_t bytes[255] = {}; + uint32_t revision = 0; + uint8_t key_len = 0; + uint8_t len = 0; + } slots[CAPACITY]; + enum class Stage : uint8_t { + Idle, TargetProbe, BackupProbe, RestoreBackup, RemoveBackup, + TempProbe, RemoveTemp, OpenWrite, ConfigureWrite, Write, CloseWrite, + OpenVerify, ConfigureVerify, VerifySize, Verify, CloseVerify, + PublishTargetProbe, PublishBackupProbe, PublishRemoveBackup, + PublishBackup, Publish, PublishCleanup, CancelClose + } stage = Stage::Idle; + File file; + uint8_t scratch[64] = {}; + uint8_t active_key[8] = {}; + uint8_t active_key_len = 0; + char target[24] = {}; + char backup[28] = {}; + uint32_t revision = 0; + uint32_t active_revision = 0; + uint32_t retry_at = 0; + uint8_t active = CAPACITY; + uint8_t next = 0; + uint8_t offset = 0; + uint8_t failures = 0; + bool target_present = false; + bool backup_present = false; + bool temp_present = false; + bool cancelling = false; + bool synchronous_io = false; + ~AdvertWriteState() { file.close(); } + void pause() { + file.close(); + stage = Stage::Idle; + active = CAPACITY; + retry_at = 0; + } + void clear() { + pause(); + for (auto& slot : slots) slot.len = 0; + failures = 0; + } + } _advert_write; + void invalidateAdvertWrite(const uint8_t key[], int key_len); + void retireAdvertWrite(const uint8_t key[], int key_len); +#endif #if !defined(NRF52_PLATFORM) bool _channel_load_incomplete = false; #if !MESH_CONTACT_CACHE @@ -163,6 +216,15 @@ public: uint8_t getBlobByKey(const uint8_t key[], int key_len, uint8_t dest_buf[]); bool putBlobByKey(const uint8_t key[], int key_len, const uint8_t src_buf[], uint8_t len); bool deleteBlobByKey(const uint8_t key[], int key_len); +#if defined(ESP32_PLATFORM) + bool queueAdvertByKey(const uint8_t key[], int key_len, + const uint8_t src_buf[], uint8_t len); + bool serviceAdvertWrites(uint32_t now); + bool flushAdvertWrites(uint32_t now = 0); + bool hasPendingAdvertWrites() const; + bool isAdvertWriteDue(uint32_t now) const; + bool consumeSynchronousAdvertIO(); +#endif File openRead(const char* filename); File openRead(FILESYSTEM* fs, const char* filename); File openDirectory(const char* path); diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 9a39c530a..53653cb83 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -6809,6 +6809,11 @@ bool MyMesh::isContactWriteDue() const { } bool MyMesh::flushContactsBeforeReboot() { +#if defined(ESP32_PLATFORM) + const uint32_t now = _ms->getMillis(); + if (_store->hasPendingAdvertWrites() && !_store->isAdvertWriteDue(now)) return false; + if (!_store->flushAdvertWrites(now)) return false; +#endif // Non-nRF stores use the legacy monolithic file and therefore do not report // dirty pages. The lazy-write deadline still proves that RAM may be newer. if (!dirty_contacts_expiry && !_store->hasPendingContactWrites()) return true; @@ -10316,6 +10321,9 @@ void MyMesh::loop() { if (held_dm_count != 0 || verified_pending_count != 0) releaseHeldOneKeyDMs(); #endif +#if defined(ESP32_PLATFORM) + servicePersistence(); +#else // is there are pending dirty contacts write needed? if (isContactWriteDue()) { const bool success = _store->serviceContactWrites(this, save_filter); @@ -10330,6 +10338,7 @@ void MyMesh::loop() { dirty_contacts_expiry, dirty_contacts_failures); } } +#endif #ifdef DISPLAY_CLASS if (_ui) { @@ -10344,6 +10353,38 @@ void MyMesh::loop() { #endif } +#if defined(ESP32_PLATFORM) +void MyMesh::servicePersistence() { + // Advert crypto runs before serial service. Defer its cache I/O until here, + // and alternate with due contact work so neither job can starve the other. + // Only one persistence job is serviced in a loop pass. + if (_store->consumeSynchronousAdvertIO()) return; + const bool contact_write_due = isContactWriteDue(); + const bool advert_write_due = _store->isAdvertWriteDue(_ms->getMillis()); + const bool service_advert = advert_write_due + && (!contact_write_due || _advert_write_next); + if (service_advert) { + _store->serviceAdvertWrites(_ms->getMillis()); + _advert_write_next = false; + return; + } + _advert_write_next = true; + if (isContactWriteDue()) { + const bool success = _store->serviceContactWrites(this, save_filter); + if (!success) { + scheduleContactWriteRetry(); + } else if (_store->hasPendingContactWrites()) { + dirty_contacts_failures = 0; + dirty_contacts_expiry = mesh::nonzeroLazyPersistenceDeadline( + futureMillis(CONTACT_PAGE_WRITE_GAP)); + } else { + mesh::resetLazyPersistenceAfterSuccess( + dirty_contacts_expiry, dirty_contacts_failures); + } + } +} +#endif + bool MyMesh::advert() { mesh::Packet* pkt; if (_prefs.advert_loc_policy == ADVERT_LOC_NONE) { @@ -10364,6 +10405,9 @@ bool MyMesh::canRecoverUsbLogging() const { // Do not discard dirty pages or interrupt a protocol/flash/radio operation. // Normal lazy writes resolve this veto without adding writes to each poll. return dirty_contacts_expiry == 0 && !_store->hasPendingContactWrites() +#if defined(ESP32_PLATFORM) + && !_store->hasPendingAdvertWrites() +#endif && !hasOutbound() && !isAnyTempRadioActive() && !hasPendingOtaApply() && !command_radio_apply_pending && !saved_radio_apply_pending && _scheduled_reboot_at == 0 && !hasPendingReqs() @@ -10418,6 +10462,9 @@ bool MyMesh::hasPendingWork() const { #endif && (!radio_apply_retry_at || millisHasNowPassed(radio_apply_retry_at))) || contact_write_needs_polling +#if defined(ESP32_PLATFORM) + || _store->hasPendingAdvertWrites() +#endif || (emergency_client_repeat_packet != NULL && millisHasNowPassed(emergency_client_repeat_send_at)) #if COMPANION_FEATURE_TEMP_RADIO diff --git a/examples/companion_radio/MyMesh.h b/examples/companion_radio/MyMesh.h index a8570ee29..d229cde71 100644 --- a/examples/companion_radio/MyMesh.h +++ b/examples/companion_radio/MyMesh.h @@ -491,7 +491,12 @@ private: return _store->getBlobByKey(key, key_len, dest_buf); } bool putBlobByKey(const uint8_t key[], int key_len, const uint8_t src_buf[], int len) override { +#if defined(ESP32_PLATFORM) + return len > 0 && len <= 255 + && _store->queueAdvertByKey(key, key_len, src_buf, len); +#else return _store->putBlobByKey(key, key_len, src_buf, len); +#endif } void checkCLIRescueCmd(); @@ -596,6 +601,9 @@ private: // helpers, short-cuts bool saveChannels() { return _store->saveChannels(this); } void saveContacts(); +#if defined(ESP32_PLATFORM) + void servicePersistence(); +#endif void scheduleContactWriteRetry(); bool isContactWriteDue() const; bool flushContactsBeforeReboot(); @@ -722,6 +730,9 @@ private: unsigned long sign_data_deadline; unsigned long dirty_contacts_expiry; uint8_t dirty_contacts_failures; +#if defined(ESP32_PLATFORM) + bool _advert_write_next = true; +#endif TransportKey send_scope; diff --git a/test/fixtures/advert_cache/FakeFilesystem.h b/test/fixtures/advert_cache/FakeFilesystem.h new file mode 100644 index 000000000..61171dfd2 --- /dev/null +++ b/test/fixtures/advert_cache/FakeFilesystem.h @@ -0,0 +1,134 @@ +#pragma once +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +class FakeFilesystem; +struct Stream { + FakeFilesystem* fs; + std::string path; + bool writer, closed = false; + size_t position = 0, buffer_size = 4096; + std::vector buffer; + Stream(FakeFilesystem*, const char*, bool); + void close(); + ~Stream() { close(); } +}; +class File { + std::shared_ptr stream; +public: + File() = default; + File(FakeFilesystem*, const char*, bool); + explicit operator bool() const { return stream && !stream->closed; } + bool setBufferSize(size_t); + size_t write(const uint8_t*, size_t); + size_t read(uint8_t*, size_t); + size_t size(); + void close() { if (stream) stream->close(); stream.reset(); } +}; +class FakeFilesystem { +public: + std::map> files; + unsigned operations = 0, handles = 0, backend_writes = 0; + unsigned rename_calls = 0, remove_calls = 0, config_calls = 0; + unsigned metadata_calls = 0, fail_metadata_at = 0; + unsigned fail_rename = 0, fail_remove = 0, fail_config = 0; + bool fail_metadata = false, fail_open = false, partial_open = false; + bool fail_read = false, short_write = false, fail_close = false, fail_format = false; + size_t max_backend_write = 0; + bool exists(const char* path) const { return files.count(path) != 0; } + bool mkdir(const char*) { ++operations; return true; } + bool format() { ++operations; if (fail_format) return false; files.clear(); return true; } + File open(const char* path, const char* mode = "r", bool = false) { + ++operations; + bool writer = mode[0] == 'w'; + if (fail_open) { + if (writer && partial_open) files[path].clear(); + return File(); + } + if (!writer && !exists(path)) return File(); + if (writer) files[path].clear(); + return File(this, path, writer); + } + bool rename(const char* from, const char* to) { + ++operations; ++rename_calls; + if (rename_calls == fail_rename || !exists(from) || exists(to)) return false; + files[to] = files.at(from); files.erase(from); return true; + } + bool remove(const char* path) { + ++operations; ++remove_calls; + assert(exists(path) && "No VFS log for normal absence"); + if (remove_calls == fail_remove) return false; + files.erase(path); return true; + } + void backend(const std::string& path, const std::vector& bytes) { + if (bytes.empty()) return; + ++backend_writes; + max_backend_write = std::max(max_backend_write, bytes.size()); + files[path].insert(files[path].end(), bytes.begin(), bytes.end()); + } +}; +inline Stream::Stream(FakeFilesystem* f, const char* p, bool w) + : fs(f), path(p), writer(w) { ++fs->handles; } +inline void Stream::close() { + if (closed) return; + ++fs->operations; + if (writer && !fs->fail_close) fs->backend(path, buffer); + buffer.clear(); closed = true; --fs->handles; +} +inline File::File(FakeFilesystem* fs, const char* path, bool writer) + : stream(std::make_shared(fs, path, writer)) {} +inline bool File::setBufferSize(size_t length) { + assert(stream && stream->position == 0); + ++stream->fs->operations; ++stream->fs->config_calls; + if (stream->fs->config_calls == stream->fs->fail_config) return false; + stream->buffer_size = length; return true; +} +inline size_t File::write(const uint8_t* data, size_t length) { + assert(stream && stream->writer); ++stream->fs->operations; + if (stream->fs->short_write && length) --length; + size_t consumed = 0; + while (consumed < length) { + if (stream->buffer.size() == stream->buffer_size) { + stream->fs->backend(stream->path, stream->buffer); stream->buffer.clear(); + } + size_t n = std::min(length-consumed, stream->buffer_size-stream->buffer.size()); + stream->buffer.insert(stream->buffer.end(), data+consumed, data+consumed+n); + consumed += n; + if (stream->buffer.size() == stream->buffer_size) { + stream->fs->backend(stream->path, stream->buffer); stream->buffer.clear(); + } + } + stream->position += length; return length; +} +inline size_t File::read(uint8_t* data, size_t length) { + assert(stream && !stream->writer); ++stream->fs->operations; + if (stream->fs->fail_read) return 0; + const auto& bytes = stream->fs->files.at(stream->path); + size_t n = std::min(length, bytes.size()-std::min(stream->position, bytes.size())); + if (n) memcpy(data, bytes.data()+stream->position, n); + stream->position += n; return n; +} +inline size_t File::size() { + assert(stream); ++stream->fs->operations; return stream->fs->files.at(stream->path).size(); +} +inline int fixtureStat(FakeFilesystem* fs, const char* path, struct stat*) { + ++fs->operations; + ++fs->metadata_calls; + if (fs->fail_metadata || fs->metadata_calls == fs->fail_metadata_at) { errno = EIO; return -1; } + assert(strncmp(path, "/spiffs", 7) == 0); + if (!fs->exists(path+7)) { errno = ENOENT; return -1; } + return 0; +} +namespace fs { using SPIFFSFS = FakeFilesystem; } +using esp_err_t = int; +constexpr int ESP_OK = 0; +inline int nvs_flash_erase() { return ESP_OK; } diff --git a/test/fixtures/advert_cache/test.cpp b/test/fixtures/advert_cache/test.cpp new file mode 100644 index 000000000..b04cc3915 --- /dev/null +++ b/test/fixtures/advert_cache/test.cpp @@ -0,0 +1,314 @@ +#include "FakeFilesystem.h" +#include +#include +#define FILESYSTEM FakeFilesystem +namespace mesh { struct Utils { + static void toHex(char* dest, const uint8_t* src, int length) { + for (int i=0; iopen(path);} + void cancelContactWrite(){} + void begin(); + bool formatFileSystem(); + bool queueAdvertByKey(const uint8_t[],int,const uint8_t[],uint8_t); + bool serviceAdvertWrites(uint32_t); + bool flushAdvertWrites(uint32_t now=0); + bool hasPendingAdvertWrites() const; + bool isAdvertWriteDue(uint32_t) const; + bool consumeSynchronousAdvertIO(); + uint8_t getBlobByKey(const uint8_t[],int,uint8_t[]); + bool putBlobByKey(const uint8_t[],int,const uint8_t[],uint8_t); + bool deleteBlobByKey(const uint8_t[],int); + bool hasPendingContactWrites() const{return contact_pending;} + bool serviceContactWrites(DataStoreHost*,bool(*)(const ContactInfo&)) { + ++_fs->operations; ++contact_passes; return true; + } + bool flushContactWrites(DataStoreHost*,bool(*)(const ContactInfo&)) { + ++_fs->operations; ++contact_passes; contact_pending=false; return true; + } +}; +#include "presence.h" +#include "store.h" +constexpr uint32_t CONTACT_PAGE_WRITE_GAP=1; +struct Clock { uint32_t now=10000; uint32_t getMillis() const{return now;} }; +struct RadioDriver { + bool isWatchdogObserving() const{return false;} + bool isCalibratingNoiseFloor() const{return false;} +} radio_driver; +struct SerialInterface { + bool hasPendingIO() const{return false;} + bool isReplyRouteAvailable(const SerialInterface*) const{return false;} +}; +struct DelayedReplies { bool hasBinaryTrace() const{return false;} }; +class MyMesh { +public: + DataStore* _store; + Clock* _ms; + bool _advert_write_next=true; + unsigned long dirty_contacts_expiry=0; + uint8_t dirty_contacts_failures=0; + bool(*save_filter)(const ContactInfo&)=nullptr; + bool _radio_available=false, _iter_started=false; + bool command_radio_apply_pending=false, saved_radio_apply_pending=false; + uint32_t _scheduled_reboot_at=0, radio_apply_retry_at=0; + uint32_t emergency_client_repeat_send_at=0; + void* sign_data=nullptr; + void* emergency_client_repeat_packet=nullptr; + SerialInterface* _serial=nullptr; + SerialInterface* _iter_reply_route=nullptr; + DelayedReplies _delayed_replies; + bool putBlobByKey(const uint8_t[],int,const uint8_t[],int); + void servicePersistence(); + bool flushContactsBeforeReboot(); + bool hasPendingWork() const; + bool canRecoverUsbLogging() const; + bool isDualRadioActive() const{return false;} + bool hasOutbound() const{return false;} + bool isAnyTempRadioActive() const{return false;} + bool hasPendingOtaApply() const{return false;} + bool hasPendingReqs() const{return false;} + bool hasQueuedWorkDue() const{return false;} + bool hasRetryWorkDue() const{return false;} + bool millisHasNowPassed(uint32_t when) const{return _ms->now>=when;} + bool isContactWriteDue() const{return _store->contact_pending;} + void scheduleContactWriteRetry(){++dirty_contacts_failures;} + unsigned long futureMillis(uint32_t delay) const{return _ms->now+delay;} +}; +#include "mesh.h" +using Key=std::array; +using Blob=std::vector; +using Stage=DataStore::AdvertWriteState::Stage; +Key key(unsigned n){Key k{};k[0]=n;return k;} +Blob blob(unsigned n, unsigned len=255){return Blob(len,n);} +std::string path(const Key& key){char path[64];makeBlobPath(key.data(),32,path,sizeof(path));return path;} +Blob get(DataStore& store,const Key& k){uint8_t bytes[255];auto n=store.getBlobByKey(k.data(),32,bytes);return Blob(bytes,bytes+n);} +void enqueue(DataStore& store,const Key& k,const Blob& b){assert(store.queueAdvertByKey(k.data(),32,b.data(),b.size()));} +void step(DataStore& store,FakeFilesystem& fs,uint32_t now=10000) { + auto before=fs.operations, backend=fs.backend_writes; + fs.max_backend_write=0; + store.serviceAdvertWrites(now); + assert(fs.operations-before<=1 && "one storage operation per cooperative pass"); + assert(fs.backend_writes-backend<=1); + assert(fs.max_backend_write<=64); +} +void drain(DataStore& store,FakeFilesystem& fs) { + unsigned n=0; + while(store.hasPendingAdvertWrites()){assert(++n<1000);step(store,fs,10000+n*1000);} + assert(fs.handles==0); +} +void until(DataStore& store,FakeFilesystem& fs,Stage target) { + unsigned n=0; + do {assert(++n<1000);step(store,fs,10000+n*1000);} + while(store._advert_write.stage!=target); +} + +void rxRouteAndScheduling() { + FakeFilesystem fs;DataStore store(fs);Clock clock;MyMesh mesh{&store,&clock}; + auto k=key(1);auto b=blob(1); + const auto before=fs.operations; + assert(mesh.putBlobByKey(k.data(),32,b.data(),b.size())); + assert(fs.operations==before && "advert_rx_no_fs"); + assert(get(store,k)==b && fs.operations==before); + store.contact_pending=true; + unsigned blob_passes=0, serial_services=0; + while(store.hasPendingAdvertWrites()) { + ++serial_services; + auto old=fs.operations, contacts=store.contact_passes; + mesh.servicePersistence(); + assert(fs.operations-old<=1); + if(store.contact_passes==contacts)++blob_passes; + ++clock.now; + } + assert(serial_services>20 && blob_passes>10 && store.contact_passes>10); + assert(fs.files.at(path(k))==b && !fs.exists("/advert.tmp")); + // Explicit durability work must not be followed by another automatic job. + enqueue(store,k,blob(2));assert(store.putBlobByKey(k.data(),32,b.data(),b.size())); + auto old=fs.operations;mesh.servicePersistence();assert(old==fs.operations); + // Disk share/export work consumes the same pass; a RAM share does not. + store.consumeSynchronousAdvertIO(); + assert(get(store,k)==b);enqueue(store,key(2),blob(2)); + old=fs.operations;mesh.servicePersistence();assert(old==fs.operations); + assert(get(store,key(2))==blob(2)); + old=fs.operations;mesh.servicePersistence();assert(fs.operations==old+1); + store.contact_pending=false;drain(store,fs); +} + +void coalesceOverflowAndDeleteEveryStage() { + for(unsigned stage=0;stage<24;++stage) { + FakeFilesystem fs;DataStore store(fs);auto k=key(1), other=key(2);auto old=blob(1), fresh=blob(2); + fs.files[path(k)]=old;enqueue(store,k,fresh); + for(unsigned i=0;iserviceContactWrites(this, save_filter)", loop) + body = function_body(mesh, "void MyMesh::servicePersistence()") self.assertIn("if (isContactWriteDue())", body) self.assertIn("_store->serviceContactWrites(this, save_filter)", body) @@ -82,7 +87,7 @@ class CompanionContactSleepContractTest(unittest.TestCase): mesh = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text() begin = function_body(mesh, "void MyMesh::begin(") save = function_body(mesh, "void MyMesh::saveContacts()") - loop = function_body(mesh, "void MyMesh::loop()") + loop = function_body(mesh, "void MyMesh::servicePersistence()") for body in (begin, save, loop): self.assertIn("mesh::nonzeroLazyPersistenceDeadline(", body) @@ -93,7 +98,7 @@ class CompanionContactSleepContractTest(unittest.TestCase): mesh = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text() header = (ROOT / "examples/companion_radio/MyMesh.h").read_text() save = function_body(mesh, "void MyMesh::saveContacts()") - loop = function_body(mesh, "void MyMesh::loop()") + loop = function_body(mesh, "void MyMesh::servicePersistence()") retry = function_body(mesh, "void MyMesh::scheduleContactWriteRetry()") self.assertIn("uint8_t dirty_contacts_failures;", header) diff --git a/test/test_esp32_advert_cache.py b/test/test_esp32_advert_cache.py new file mode 100644 index 000000000..06e5015a2 --- /dev/null +++ b/test/test_esp32_advert_cache.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python3 +"""Exercise production ESP advert jobs and their MyMesh route under sanitizers.""" +from pathlib import Path +import subprocess +import tempfile +import unittest +from test_t096_full_memory import method + +ROOT = Path(__file__).resolve().parents[1] + + +class AdvertCacheTest(unittest.TestCase): + def test_inline_contacts(self): + self.run_fixture(0) + + def test_cached_contacts(self): + self.run_fixture(1) + + def test_synchronous_received_advert_is_detected(self): + self.run_fixture(0, synchronous=True) + + def run_fixture(self, cache, synchronous=False): + source = (ROOT / 'examples/companion_radio/DataStore.cpp').read_text() + header = (ROOT / 'examples/companion_radio/DataStore.h').read_text() + start = header.index(' struct AdvertWriteState {') + end = header.index('\n#endif', start) + state = header[start:end] + signatures = ('static File openWrite(', 'inline void makeBlobPath(', 'bool DataStore::queueAdvertByKey(', + 'void DataStore::retireAdvertWrite(', 'void DataStore::invalidateAdvertWrite(', + 'bool DataStore::consumeSynchronousAdvertIO(', + 'bool DataStore::hasPendingAdvertWrites() const', + 'bool DataStore::isAdvertWriteDue(', 'bool DataStore::serviceAdvertWrites(', + 'bool DataStore::flushAdvertWrites(', 'void DataStore::begin()', + 'bool DataStore::formatFileSystem()') + # ESP/non-LittleFS blob methods follow the final makeBlobPath helper. + blob_source = source[source.rindex('inline void makeBlobPath('):] + implementation = '\n'.join(method(source, s) for s in signatures) + implementation += '\n' + '\n'.join(method(blob_source, s) for s in ( + 'uint8_t DataStore::getBlobByKey(', 'bool DataStore::putBlobByKey(', + 'bool DataStore::deleteBlobByKey(')) + presence = method(source, 'static bool companionPathPresence(').replace( + '::stat(vfs_path, &info)', 'fixtureStat(fs, vfs_path, &info)') + mesh_header = (ROOT / 'examples/companion_radio/MyMesh.h').read_text() + route = method(mesh_header, ' bool putBlobByKey(').replace(' override', '') + route = route.replace('bool putBlobByKey(', 'bool MyMesh::putBlobByKey(', 1) + if synchronous: + route = route.replace('_store->queueAdvertByKey(', '_store->putBlobByKey(') + mesh_source = (ROOT / 'examples/companion_radio/MyMesh.cpp').read_text() + scheduling = method(mesh_source, 'void MyMesh::servicePersistence()') + flush = method(mesh_source, 'bool MyMesh::flushContactsBeforeReboot()') + work = method(mesh_source, 'bool MyMesh::hasPendingWork() const') + recovery = method(mesh_source, 'bool MyMesh::canRecoverUsbLogging() const') + # The entire production scheduling body runs below. This order check + # additionally pins its call site after either serial dispatch route. + loop = method(mesh_source, 'void MyMesh::loop()') + self.assertLess(loop.index('checkSerialInterface();'), loop.index('servicePersistence();')) + self.assertLess(loop.index('checkCLIRescueCmd();'), loop.index('servicePersistence();')) + with tempfile.TemporaryDirectory(prefix='mesh-advert-jobs-') as directory: + temp = Path(directory) + (temp / 'advert_state.h').write_text(state) + (temp / 'presence.h').write_text(presence) + (temp / 'store.h').write_text(implementation) + (temp / 'mesh.h').write_text('\n'.join((route, scheduling, flush, work, recovery))) + binary = temp / 'test' + result = subprocess.run([ + 'c++', '-std=c++17', '-O1', '-g', '-Wall', '-Wextra', + '-fsanitize=address,undefined', '-fno-omit-frame-pointer', + '-DESP32_PLATFORM=1', '-DESP32=1', f'-DMESH_CONTACT_CACHE={cache}', + '-I', str(ROOT / 'test/fixtures/advert_cache'), + '-I', str(ROOT / 'src'), '-I', str(temp), + str(ROOT / 'test/fixtures/advert_cache/test.cpp'), '-o', str(binary), + ], capture_output=True, text=True) + self.assertEqual(result.returncode, 0, result.stderr) + result = subprocess.run([str(binary)], capture_output=True, text=True) + if synchronous: + self.assertNotEqual(result.returncode, 0) + self.assertIn('advert_rx_no_fs', result.stderr) + else: + self.assertEqual(result.returncode, 0, result.stdout+result.stderr) + + +if __name__ == '__main__': + unittest.main()