diff --git a/src/helpers/ContactFileTransaction.h b/src/helpers/ContactFileTransaction.h index 6ecf48fe..6ddec2ca 100644 --- a/src/helpers/ContactFileTransaction.h +++ b/src/helpers/ContactFileTransaction.h @@ -64,6 +64,12 @@ public: _file = fs->open(_temp, "w", true); #endif _ok = static_cast(_file); +#if defined(ESP32_PLATFORM) + // Arduino's default 4 KiB stdio buffer hides large SPIFFS writes inside a + // single small write() pass. Configure this newly opened stream before + // its first I/O, keeping one contact record as the largest buffered batch. + _ok = _ok && _file.setBufferSize(storage::CONTACT_RECORD_SIZE); +#endif } ~ContactFileTransaction() { if (_file) _file.close(); @@ -95,7 +101,13 @@ public: } if (ok && _commit_stage == CommitStage::Open) { _verify = _fs->open(_temp, "r"); - ok = _verify && _verify.size() == _size; + ok = static_cast(_verify); +#if defined(ESP32_PLATFORM) + // fread() otherwise reads ahead up to 4 KiB while the caller requests + // only one 64-byte CRC chunk. A failed configuration cannot publish. + ok = ok && _verify.setBufferSize(64); +#endif + ok = ok && _verify.size() == _size; if (ok) { _verify_remaining = _size; _commit_stage = CommitStage::Verify; diff --git a/test/fixtures/companion_uncached_storage/test.cpp b/test/fixtures/companion_uncached_storage/test.cpp index 6ecccea8..7781ae9a 100644 --- a/test/fixtures/companion_uncached_storage/test.cpp +++ b/test/fixtures/companion_uncached_storage/test.cpp @@ -28,6 +28,7 @@ public: size_t write(const uint8_t* bytes, size_t length); size_t size() const; int available() const { return static_cast(size() - position); } + bool setBufferSize(size_t size) { return size != 0; } void flush() {} void close() { fs = nullptr; } }; diff --git a/test/fixtures/contact_cache/FakeFilesystem.h b/test/fixtures/contact_cache/FakeFilesystem.h index 73087d6a..a17385d5 100644 --- a/test/fixtures/contact_cache/FakeFilesystem.h +++ b/test/fixtures/contact_cache/FakeFilesystem.h @@ -20,6 +20,10 @@ class File { bool _write = false; unsigned _read_generation = 0; std::vector _read_inode; + size_t _stdio_size = 4096, _backend_position = 0; + std::vector _stdio_buffer; + size_t _read_buffer_position = 0; + void flushWriteBuffer(); public: File() = default; explicit File(FakeFilesystem& fs) : _fs(&fs) {} @@ -30,8 +34,9 @@ public: size_t write(const uint8_t* bytes, size_t length); size_t size() const; bool seek(size_t position); - void flush() {} - void close() { _fs = nullptr; } + bool setBufferSize(size_t size); + void flush(); + void close(); }; class FakeFilesystem { @@ -43,6 +48,13 @@ public: size_t capacity = 1024 * 1024; size_t max_write = std::numeric_limits::max(); size_t largest_read = 0, largest_write = 0; + // ESP's File wrapper is a buffered FILE*. Enable its 4 KiB default in + // production transaction tests, and count backend I/O separately from + // fwrite/fread calls so logical record/chunk bounds cannot hide batching. + bool emulate_stdio = false; + size_t largest_backend_read = 0, largest_backend_write = 0; + unsigned backend_reads = 0, backend_writes = 0; + unsigned buffer_config_calls = 0, fail_buffer_config = 0; std::string fail_read; bool metadata_error = false, fail_create = false, fail_remove = false; unsigned removes = 0, missing_remove_logs = 0; @@ -116,6 +128,29 @@ size_t File::read(uint8_t* bytes, size_t length) { // 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; + if (_fs->emulate_stdio) { + size_t copied = 0; + while (copied < length) { + if (_read_buffer_position == _stdio_buffer.size()) { + const size_t available = data.size() - std::min(data.size(), _backend_position); + const size_t count = std::min(_stdio_size, available); + if (count == 0) break; + ++_fs->backend_reads; + _fs->largest_backend_read = std::max(_fs->largest_backend_read, count); + _stdio_buffer.assign(data.begin() + _backend_position, + data.begin() + _backend_position + count); + _backend_position += count; + _read_buffer_position = 0; + } + const size_t count = std::min(length - copied, + _stdio_buffer.size() - _read_buffer_position); + memcpy(bytes + copied, _stdio_buffer.data() + _read_buffer_position, count); + _read_buffer_position += count; + copied += count; + } + _position += copied; + return copied; + } const size_t count = std::min(length, data.size() - std::min(data.size(), _position)); if (count) memcpy(bytes, data.data() + _position, count); _position += count; @@ -126,6 +161,17 @@ size_t File::write(const uint8_t* bytes, size_t length) { ++_fs->writes; _fs->largest_write = std::max(_fs->largest_write, length); const size_t count = std::min(length, _fs->max_write); + if (_fs->emulate_stdio) { + size_t consumed = 0; + while (consumed < count) { + if (_stdio_buffer.size() == _stdio_size) flushWriteBuffer(); + const size_t part = std::min(count - consumed, _stdio_size - _stdio_buffer.size()); + _stdio_buffer.insert(_stdio_buffer.end(), bytes + consumed, bytes + consumed + part); + consumed += part; + } + _position += count; + return count; + } auto& data = _fs->files[_path]; data.resize(_position + count); memcpy(data.data() + _position, bytes, count); @@ -140,5 +186,27 @@ size_t File::size() const { bool File::seek(size_t pos) { if (!_fs || pos > size()) return false; _position = pos; + _backend_position = pos; + _stdio_buffer.clear(); + _read_buffer_position = 0; return true; } +bool File::setBufferSize(size_t size) { + assert(_position == 0 && _stdio_buffer.empty()); // before the stream's first I/O + if (!_fs || size == 0) return false; + if (++_fs->buffer_config_calls == _fs->fail_buffer_config) return false; + _stdio_size = size; + return true; +} +void File::flushWriteBuffer() { + if (!_fs || !_write || _stdio_buffer.empty()) return; + ++_fs->backend_writes; + _fs->largest_backend_write = std::max(_fs->largest_backend_write, _stdio_buffer.size()); + auto& data = _fs->files[_path]; + data.resize(_backend_position + _stdio_buffer.size()); + memcpy(data.data() + _backend_position, _stdio_buffer.data(), _stdio_buffer.size()); + _backend_position += _stdio_buffer.size(); + _stdio_buffer.clear(); +} +void File::flush() { if (_fs && _fs->emulate_stdio && _write) flushWriteBuffer(); } +void File::close() { flush(); _fs = nullptr; } diff --git a/test/fixtures/contact_persistence/test_esp32_uncached.cpp b/test/fixtures/contact_persistence/test_esp32_uncached.cpp index 51ebe5dd..a5e40664 100644 --- a/test/fixtures/contact_persistence/test_esp32_uncached.cpp +++ b/test/fixtures/contact_persistence/test_esp32_uncached.cpp @@ -115,10 +115,13 @@ unsigned drain(DataStore& store, Host& h, Filter filter = nullptr) { unsigned passes = 0; while (store.hasPendingContactWrites()) { const unsigned writes = SPIFFS.writes, reads = SPIFFS.reads; + const unsigned backend_writes = SPIFFS.backend_writes, backend_reads = SPIFFS.backend_reads; assert(++passes < 5000); assert(store.serviceContactWrites(&h, filter)); assert(SPIFFS.writes - writes <= 1); assert(SPIFFS.reads - reads <= 1); + assert(SPIFFS.backend_writes - backend_writes <= 1); + assert(SPIFFS.backend_reads - backend_reads <= 1); // Transport service runs here, between every production pass. It must // never observe an unpublished replacement or a target-name gap. if (store.hasPendingContactWrites()) { @@ -143,6 +146,7 @@ void bootCheck(const Host& expected, Filter filter = nullptr) { void boundedFullTableAndFirstSave() { SPIFFS = FakeFilesystem(); + SPIFFS.emulate_stdio = true; Host h = table(350); for (unsigned i = 0; i < h.contacts.size(); i += 7) h.contacts[i].type = 0; DataStore store(SPIFFS); @@ -151,6 +155,8 @@ void boundedFullTableAndFirstSave() { const auto passes = drain(store, h, keepContact); assert(passes > expected.size() / 64); assert(SPIFFS.largest_read <= 64); + assert(SPIFFS.largest_backend_read <= 64); + assert(SPIFFS.largest_backend_write <= mesh::storage::CONTACT_RECORD_SIZE); assert(SPIFFS.largest_write == mesh::storage::CONTACT_RECORD_SIZE); assert(SPIFFS.removes == 0); // first publication has no backup/temp to delete assert(h.copies >= h.contacts.size()); @@ -159,6 +165,7 @@ void boundedFullTableAndFirstSave() { void mutationRestart(unsigned phase) { SPIFFS = FakeFilesystem(); + SPIFFS.emulate_stdio = true; Host h = table(); SPIFFS.files["/contacts3"] = image(h); const auto old = SPIFFS.files.at("/contacts3"); @@ -187,6 +194,7 @@ void mutationRestart(unsigned phase) { void faultAndRetry(unsigned fault) { SPIFFS = FakeFilesystem(); + SPIFFS.emulate_stdio = true; Host h = table(); SPIFFS.files["/contacts3"] = image(h); const auto old = SPIFFS.files.at("/contacts3"); @@ -196,10 +204,12 @@ void faultAndRetry(unsigned fault) { if (fault == 0) SPIFFS.max_write = 50; if (fault == 1) SPIFFS.fail_create = true; if (fault == 7) SPIFFS.metadata_error = true; - bool injected = fault == 0 || fault == 1 || fault == 7; + if (fault == 8 || fault == 9) SPIFFS.fail_buffer_config = fault - 7; + bool injected = fault == 0 || fault == 1 || fault >= 7; bool success = true; for (unsigned pass = 0; pass < 1000 && success; ++pass) { - if (!injected && store.verifying()) { + if (!injected && store.verifying() + && SPIFFS.files.at("/contacts3.tmp").size() == image(h).size()) { if (fault == 2) SPIFFS.fail_read = "/contacts3.tmp"; if (fault == 3) SPIFFS.files.at("/contacts3.tmp")[20] ^= 1; if (fault == 4) SPIFFS.files.at("/contacts3.tmp").pop_back(); @@ -220,12 +230,14 @@ void faultAndRetry(unsigned fault) { SPIFFS.fail_create = SPIFFS.metadata_error = false; SPIFFS.fail_read.clear(); SPIFFS.fail_rename = 0; + SPIFFS.fail_buffer_config = 0; drain(store, h); bootCheck(h); } void syncFlushAndFilterChange() { SPIFFS = FakeFilesystem(); + SPIFFS.emulate_stdio = true; Host h = table(); SPIFFS.files["/contacts3"] = image(h); DataStore store(SPIFFS); @@ -254,6 +266,7 @@ void syncFlushAndFilterChange() { void realBackupCleanupFailure() { SPIFFS = FakeFilesystem(); + SPIFFS.emulate_stdio = true; Host h = table(); const auto old = image(h); SPIFFS.files["/contacts3"] = old; @@ -278,6 +291,7 @@ void realBackupCleanupFailure() { void rebootAbortRecoveryAndDurableRemoval() { SPIFFS = FakeFilesystem(); + SPIFFS.emulate_stdio = true; Host h = table(); const auto old = image(h); SPIFFS.files["/contacts3"] = old; @@ -312,6 +326,7 @@ void rebootAbortRecoveryAndDurableRemoval() { void expectedAbsenceAndRealDeleteFailures() { SPIFFS = FakeFilesystem(); + SPIFFS.emulate_stdio = true; DataStore store(SPIFFS); uint8_t key[32] = {1, 2, 3}; char path[64]; @@ -345,7 +360,7 @@ void expectedAbsenceAndRealDeleteFailures() { int main() { boundedFullTableAndFirstSave(); for (unsigned phase = 0; phase < 3; ++phase) mutationRestart(phase); - for (unsigned fault = 0; fault < 8; ++fault) faultAndRetry(fault); + for (unsigned fault = 0; fault < 10; ++fault) faultAndRetry(fault); syncFlushAndFilterChange(); realBackupCleanupFailure(); rebootAbortRecoveryAndDurableRemoval(); diff --git a/test/fixtures/radio_profiles/mocks/helpers/IdentityStore.h b/test/fixtures/radio_profiles/mocks/helpers/IdentityStore.h index bf3127f8..13456fbf 100644 --- a/test/fixtures/radio_profiles/mocks/helpers/IdentityStore.h +++ b/test/fixtures/radio_profiles/mocks/helpers/IdentityStore.h @@ -22,6 +22,7 @@ class File { bool isDirectory() const { return false; } int read(uint8_t* data, size_t size); size_t write(const uint8_t* data, size_t size); + bool setBufferSize(size_t size) { return size != 0; } void flush() {} void close() { fs_ = nullptr; } }; diff --git a/test/fixtures/telemetry_tx_persistence/IdentityStore.h b/test/fixtures/telemetry_tx_persistence/IdentityStore.h index 76f76527..0db2a923 100644 --- a/test/fixtures/telemetry_tx_persistence/IdentityStore.h +++ b/test/fixtures/telemetry_tx_persistence/IdentityStore.h @@ -26,6 +26,7 @@ public: bool isDirectory() const { return false; } int read(uint8_t* data, size_t size); size_t write(const uint8_t* data, size_t size); + bool setBufferSize(size_t size) { return fs_ != nullptr && size != 0; } void flush() {} void close() { fs_ = nullptr; } }; diff --git a/test/test_esp32_cooperative_contacts.py b/test/test_esp32_cooperative_contacts.py index 3cad31c3..6acb7bdb 100644 --- a/test/test_esp32_cooperative_contacts.py +++ b/test/test_esp32_cooperative_contacts.py @@ -16,7 +16,13 @@ class UncachedESPContactsTest(unittest.TestCase): def test_explicit_uncached_qualification_configuration(self): self.run_fixture(["-DMESH_CONTACT_CACHE=0"]) - def run_fixture(self, policy): + def test_default_stdio_writer_batch_is_detected(self): + self.run_fixture(["-DBOARD_HAS_PSRAM=1"], disable="writer") + + def test_default_stdio_reader_readahead_is_detected(self): + self.run_fixture(["-DBOARD_HAS_PSRAM=1"], disable="reader") + + def run_fixture(self, policy, disable=None): with tempfile.TemporaryDirectory(prefix="mesh-esp-inline-contacts-") as directory: temp = Path(directory) source = (ROOT / "examples/companion_radio/DataStore.cpp").read_text() @@ -56,6 +62,25 @@ class UncachedESPContactsTest(unittest.TestCase): implementation += 'static const char hex_chars[] = "0123456789ABCDEF";\n' implementation += method(utils, "void Utils::toHex(") + "\n}\n" (temp / "store_under_test.h").write_text(implementation) + transaction = (ROOT / "src/helpers/ContactFileTransaction.h").read_text() + if disable == "writer": + self.assertIn("_ok = _ok && _file.setBufferSize(storage::CONTACT_RECORD_SIZE);", transaction) + transaction = transaction.replace( + "_ok = _ok && _file.setBufferSize(storage::CONTACT_RECORD_SIZE);", + "// Negative control: retain the SDK's default writer buffer.") + elif disable == "reader": + self.assertIn("ok = ok && _verify.setBufferSize(64);", transaction) + transaction = transaction.replace("ok = ok && _verify.setBufferSize(64);", + "// Negative control: retain the SDK's default reader buffer.") + transaction = transaction.replace('#include "IdentityStore.h"', + '#include ') + transaction = transaction.replace('#include "PersistentStoreFormat.h"', + '#include ') + (temp / "transaction_under_test.h").write_text(transaction) + fixture = (ROOT / "test/fixtures/contact_persistence/test_esp32_uncached.cpp").read_text() + fixture = fixture.replace('#include ', + '#include "transaction_under_test.h"') + (temp / "test.cpp").write_text(fixture) packet = (ROOT / "src/Packet.cpp").read_text() (temp / "packet_under_test.h").write_text("namespace mesh {\n" + "\n".join( method(packet, signature) for signature in ( @@ -71,12 +96,17 @@ class UncachedESPContactsTest(unittest.TestCase): "-I", str(ROOT / "test/fixtures/contact_cache"), "-I", str(ROOT / "lib/ed25519"), "-I", str(ROOT / "src"), "-I", str(temp), - str(ROOT / "test/fixtures/contact_persistence/test_esp32_uncached.cpp"), + str(temp / "test.cpp"), "-lcrypto", "-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) - self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + if disable is None: + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + else: + self.assertNotEqual(result.returncode, 0) + self.assertIn("largest_backend_" + ("write" if disable == "writer" else "read"), + result.stderr) if __name__ == "__main__":