FileTelemetryStore: order the ring by record sequence, not by the header

A full ring overwrites its oldest slot in place, so the header commit that
followed was the only thing making that write consistent. If it failed, the
in-file header described a ring the file no longer held, and at(0) returned
the replacement record as the oldest, before and after a reboot.

Records now carry a sequence number, which doubles as the occupied flag, and
head/count/lastSeq are derived by scanning the slots on open. The header is
written once at create and never again, so a push is a single record write
with nothing left to commit afterwards. That also halves the writes per
reading, which matters on the flash-backed variants.

Also assert the truncating write in the geometry test, which would otherwise
pass on an empty file for the wrong reason.

src/modules/Telemetry/FileTelemetryStore.h

@@ -23,20 +23,20 @@ template <typename T, typename FsT = typename std::remove_reference<decltype(FSC
 class FileTelemetryStore : public TelemetryStore<T>
 {
     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<T>
     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<T>
         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<T>
             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<T>
         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<T>
         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<T>
             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<T>
         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<T> &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<T>
         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();

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<uint32_t> 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<uint32_t> 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<uint32_t> s(kStorePath, 4);
+    TelemetryReading<uint32_t> 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());