diff --git a/src/modules/Telemetry/FileTelemetryStore.h b/src/modules/Telemetry/FileTelemetryStore.h index 28a1b17472..d6959ab536 100644 --- a/src/modules/Telemetry/FileTelemetryStore.h +++ b/src/modules/Telemetry/FileTelemetryStore.h @@ -23,20 +23,20 @@ template { static constexpr uint32_t MAGIC = 0x4D544853; // "MTHS" - static constexpr uint16_t VERSION = 1; + static constexpr uint16_t VERSION = 2; - // Packed: its layout has to be stable to reopen a file. + // Packed, and immutable once written: its layout has to be stable to reopen a file. struct __attribute__((packed)) Header { uint32_t magic; uint16_t version; uint16_t recordSize; // catches payload-layout drift across firmware builds uint16_t slots; - uint16_t head; - uint16_t count; }; // Not packed: T holds floats and unaligned access faults. Drift is caught by hdr.recordSize. + // seq orders the ring and doubles as the occupied flag, so it is never handed out as 0. struct Record { + uint32_t seq; uint32_t time; uint8_t publishedMask; T metrics; @@ -47,6 +47,13 @@ class FileTelemetryStore : public TelemetryStore Header hdr = {}; bool usable = false; + // Derived from the records on open, never persisted on its own. A ring whose order lives in the + // header needs that header committed after every push, and a push that overwrote the oldest slot + // has nothing to roll back to if the commit fails. + uint16_t head = 0; + uint16_t count = 0; + uint32_t lastSeq = 0; + static uint32_t offsetOf(uint16_t slot) { return sizeof(Header) + (uint32_t)slot * sizeof(Record); } bool fileExists() @@ -66,14 +73,26 @@ class FileTelemetryStore : public TelemetryStore return n; } - bool writeHeader(const Header &h) + bool readSlot(uint16_t slot, Record &r) + { + concurrency::LockGuard g(spiLock); + auto f = fs.open(path, FILE_O_READ); + if (!f) + return false; + f.seek(offsetOf(slot)); + const bool ok = f.read((uint8_t *)&r, sizeof(r)) == sizeof(r); + f.close(); + return ok; + } + + bool writeSlot(uint16_t slot, const Record &r) { concurrency::LockGuard g(spiLock); auto f = fs.open(path, TELEMETRY_STORE_O_RW); if (!f) return false; - f.seek(0); - const bool ok = f.write((const uint8_t *)&h, sizeof(h)) == sizeof(h); + f.seek(offsetOf(slot)); + const bool ok = f.write((const uint8_t *)&r, sizeof(r)) == sizeof(r); f.flush(); f.close(); return ok; @@ -93,7 +112,7 @@ class FileTelemetryStore : public TelemetryStore return false; } - hdr = {MAGIC, VERSION, (uint16_t)sizeof(Record), slots, 0, 0}; + hdr = {MAGIC, VERSION, (uint16_t)sizeof(Record), slots}; bool ok = f.write((const uint8_t *)&hdr, sizeof(hdr)) == sizeof(hdr); const Record blank = {}; @@ -103,6 +122,9 @@ class FileTelemetryStore : public TelemetryStore f.flush(); f.close(); + head = count = 0; + lastSeq = 0; + if (!ok) LOG_ERROR("Telemetry store: cannot preallocate %s, %u slots", path, (unsigned)slots); return ok; @@ -119,6 +141,28 @@ class FileTelemetryStore : public TelemetryStore return ok; } + /// Rebuild ring order from the records themselves; the oldest is the lowest sequence present. + void scanSlots() + { + uint32_t oldestSeq = UINT32_MAX; + head = count = 0; + lastSeq = 0; + + for (uint16_t i = 0; i < hdr.slots; i++) { + Record r = {}; + if (!readSlot(i, r) || r.seq == 0) + continue; + + count++; + if (r.seq > lastSeq) + lastSeq = r.seq; + if (r.seq < oldestSeq) { + oldestSeq = r.seq; + head = i; + } + } + } + public: /// @param fs filesystem to keep it on; SD, PSRamFS, anything with the same open/exists subset. FileTelemetryStore(const char *path, uint16_t slots, FsT &fs = FSCom) : fs(fs), path(path) @@ -135,16 +179,17 @@ class FileTelemetryStore : public TelemetryStore return; } - // Inconsistent header, or a file shorter than the geometry it claims: an interrupted - // preallocation would leave the second, and at() would read past the end - if (hdr.head >= slots || hdr.count > slots || fileSize() < offsetOf(slots)) { - LOG_WARN("Telemetry store: %s geometry is inconsistent, recreating", path); + // Shorter than the geometry its header claims: an interrupted preallocation leaves that, + // and readSlot() would run off the end + if (fileSize() < offsetOf(slots)) { + LOG_WARN("Telemetry store: %s is short of its geometry, recreating", path); usable = create(slots); return; } usable = true; - LOG_INFO("Telemetry store: %s reopened, %u/%u readings", path, (unsigned)hdr.count, (unsigned)hdr.slots); + scanSlots(); + LOG_INFO("Telemetry store: %s reopened, %u/%u readings", path, (unsigned)count, (unsigned)hdr.slots); } FileTelemetryStore(const FileTelemetryStore &) = delete; @@ -155,59 +200,38 @@ class FileTelemetryStore : public TelemetryStore if (!usable) return false; - const uint16_t slot = (hdr.head + hdr.count) % hdr.slots; - Record r = {}; + r.seq = lastSeq + 1; r.time = time; r.publishedMask = 0; r.metrics = metrics; - { - concurrency::LockGuard g(spiLock); - auto f = fs.open(path, TELEMETRY_STORE_O_RW); - if (!f) - return false; - f.seek(offsetOf(slot)); - const bool ok = f.write((const uint8_t *)&r, sizeof(r)) == sizeof(r); - f.flush(); - f.close(); - if (!ok) - return false; - } - - Header next = hdr; - if (next.count < next.slots) - next.count++; - else - next.head = (next.head + 1) % next.slots; // the write above overwrote the old head - - // Commit in RAM only once it is on disk, so a failed write cannot expose an uncounted slot - if (!writeHeader(next)) + const uint16_t slot = (head + count) % hdr.slots; + if (!writeSlot(slot, r)) return false; - hdr = next; + + // The record carries its own order, so this is bookkeeping, not a second commit that could + // fail and leave the file describing a ring it no longer holds + lastSeq = r.seq; + if (count < hdr.slots) + count++; + else + head = (head + 1) % hdr.slots; // the write above overwrote the old head + return true; } - uint16_t size() const override { return usable ? hdr.count : 0; } + uint16_t size() const override { return usable ? count : 0; } uint16_t capacity() const override { return usable ? hdr.slots : 0; } bool at(uint16_t i, TelemetryReading &out) override { - if (!usable || i >= hdr.count) + if (!usable || i >= count) return false; Record r = {}; - { - concurrency::LockGuard g(spiLock); - auto f = fs.open(path, FILE_O_READ); - if (!f) - return false; - f.seek(offsetOf((hdr.head + i) % hdr.slots)); - const bool ok = f.read((uint8_t *)&r, sizeof(r)) == sizeof(r); - f.close(); - if (!ok) - return false; - } + if (!readSlot((head + i) % hdr.slots, r)) + return false; out.metrics = r.metrics; out.time = r.time; @@ -228,7 +252,7 @@ class FileTelemetryStore : public TelemetryStore if (!f) return; // Just the mask byte; the metrics beside it are unchanged - f.seek(offsetOf((hdr.head + i) % hdr.slots) + offsetof(Record, publishedMask)); + f.seek(offsetOf((head + i) % hdr.slots) + offsetof(Record, publishedMask)); const bool ok = f.write(&mask, 1) == 1; f.flush(); f.close(); diff --git a/test/test_air_quality_telemetry/test_main.cpp b/test/test_air_quality_telemetry/test_main.cpp index fbc3e53931..19fc30705a 100644 --- a/test/test_air_quality_telemetry/test_main.cpp +++ b/test/test_air_quality_telemetry/test_main.cpp @@ -259,9 +259,9 @@ static void test_store_fileRebuiltWhenTruncated() TEST_ASSERT_TRUE(s.push(11U, 1000U)); } - // Keep the header, drop every slot behind it. 14 is the packed Header: magic, version, - // recordSize, slots, head, count. - uint8_t header[14]; + // Keep the header, drop every slot behind it. 10 is the packed Header: magic, version, + // recordSize, slots. + uint8_t header[10]; File r = FSCom.open(kStorePath, FILE_O_READ); TEST_ASSERT_TRUE(r); TEST_ASSERT_EQUAL_INT(sizeof(header), r.read(header, sizeof(header))); @@ -269,7 +269,8 @@ static void test_store_fileRebuiltWhenTruncated() File w = FSCom.open(kStorePath, FILE_O_WRITE); // truncates TEST_ASSERT_TRUE(w); - w.write(header, sizeof(header)); + // Assert it, or a failed write leaves an empty file that rebuilds for the wrong reason + TEST_ASSERT_EQUAL_INT(sizeof(header), w.write(header, sizeof(header))); w.close(); FileTelemetryStore s(kStorePath, 6); @@ -277,6 +278,36 @@ static void test_store_fileRebuiltWhenTruncated() TEST_ASSERT_TRUE(s.isEmpty()); } +// Ring order is derived from the records, so a wrapped ring reopens in the same order it had, with +// no separately committed header to disagree with them. +static void test_store_fileKeepsOrderAcrossReopenWhenFull() +{ + freshStoreFile(); + { + FileTelemetryStore s(kStorePath, 4); + for (uint32_t i = 1; i <= 6U; i++) // two past capacity, so head is off zero + TEST_ASSERT_TRUE(s.push(100U + i, 2000U + i)); + } + + FileTelemetryStore s(kStorePath, 4); + TelemetryReading r; + TEST_ASSERT_TRUE(s.isUsable()); + TEST_ASSERT_EQUAL_UINT16(4, s.size()); + for (uint16_t i = 0; i < 4; i++) { + TEST_ASSERT_TRUE(s.at(i, r)); + TEST_ASSERT_EQUAL_UINT32(103U + i, r.metrics); + TEST_ASSERT_EQUAL_UINT32(2003U + i, r.time); + } + + // And a push after reopening continues the ring rather than restarting it + TEST_ASSERT_TRUE(s.push(999U, 3000U)); + TEST_ASSERT_EQUAL_UINT16(4, s.size()); + TEST_ASSERT_TRUE(s.at(0, r)); + TEST_ASSERT_EQUAL_UINT32(104U, r.metrics); + TEST_ASSERT_TRUE(s.newest(r)); + TEST_ASSERT_EQUAL_UINT32(999U, r.metrics); +} + // Preallocated at creation, so a full store costs the same as an empty one and cannot fill the // filesystem later. static void test_store_fileDoesNotGrowWithUse() @@ -340,6 +371,7 @@ void setup() RUN_TEST(test_store_fileSurvivesReopen); RUN_TEST(test_store_fileRebuiltWhenGeometryChanges); RUN_TEST(test_store_fileRebuiltWhenTruncated); + RUN_TEST(test_store_fileKeepsOrderAcrossReopenWhenFull); RUN_TEST(test_store_fileDoesNotGrowWithUse); #endif exit(UNITY_END());