oura: extract the history-drain/resume-cursor decisions into a pure, tested core (#291 follow-up)

Packages/OuraProtocol/Sources/OuraProtocol/OuraHistoryDrain.swift

@@ -0,0 +1,86 @@
+import Foundation
+
+/// Pure, testable decision core for the Oura history drain + durable resume cursor (#91 / #291).
+///
+/// Extracted from `OuraLiveSource` so the drain guards and cursor logic — which silently regressed once
+/// already when the Oura BLE stack was refactored (#291: "lost when refactoring") — are pinned by unit
+/// tests instead of only on-device observation. This owns ONLY the per-drain counters and the decisions;
+/// the caller (`OuraLiveSource`) keeps all I/O — anchor resolution, persistence, logging, and the actual
+/// `historyCursorAdvanced` emit.
+///
+/// The `0x11` history summary carries no cursor — only `bytes_left` (a remaining-byte count) and a
+/// `moreData` flag. A healthy drain runs until `bytes_left == 0`; the resume cursor commits from the
+/// newest STORED sample's ring-time at drain end. Byte counts are never persisted.
+public struct OuraHistoryDrain: Sendable, Equatable {
+    /// A ring-time above this is corrupt (~1.6 years of ticks) and must not set the resume cursor, or the
+    /// next session would seek into nonsense. Bounds the cursor at the source.
+    public static let maxPlausibleResumeTicks: UInt32 = 500_000_000
+    /// `bytes_left` not shrinking for this many straight summaries = the ring is looping; stop.
+    public static let maxStallSummaries = 3
+    /// A healthy full pull finishes in ~1-2 min; past this something upstream is wrong — stop, keep the
+    /// banked progress, and let the periodic re-fetch try again later.
+    public static let maxDrainSeconds: TimeInterval = 300
+
+    private var minBytesLeftSeen: UInt32 = .max
+    private var stallCount = 0
+    /// Newest STORED ring-time seen this drain — the value the resume cursor commits to at drain end.
+    public private(set) var maxStoredRingTime: UInt32 = 0
+    /// A real stored sample older than where we sought this fetch: the ring's clock reset (or it ignored
+    /// the seek), so the persisted cursor is stale → full pull next connect.
+    public private(set) var sawPreResumeData = false
+
+    public init() {}
+
+    /// Reset the per-drain state at the start of a fetch. Mirrors `OuraLiveSource`'s fetch-start reset.
+    public mutating func reset() {
+        minBytesLeftSeen = .max
+        stallCount = 0
+        maxStoredRingTime = 0
+        sawPreResumeData = false
+    }
+
+    /// Fold in one history summary; returns whether the drain should CONTINUE.
+    ///
+    /// `moreData == false` (`bytes_left == 0`) always completes. Otherwise two backstops force-stop a
+    /// misbehaving ring while keeping banked progress: the STALL guard (`bytes_left` must shrink across
+    /// summaries — [maxStallSummaries] flat reads means it's looping) and the DEADLINE guard
+    /// (`elapsedSeconds` past [maxDrainSeconds]). `elapsedSeconds` is the caller's wall clock since the
+    /// drain started, or `0` when no drain-start is set (matching the old `let started = …` guard).
+    public mutating func onSummary(bytesLeft: UInt32, moreData: Bool, elapsedSeconds: TimeInterval) -> Bool {
+        guard moreData else { return false }
+        if bytesLeft < minBytesLeftSeen {
+            minBytesLeftSeen = bytesLeft
+            stallCount = 0
+        } else {
+            stallCount += 1
+            if stallCount >= Self.maxStallSummaries { return false }
+        }
+        if elapsedSeconds > Self.maxDrainSeconds { return false }
+        return true
+    }
+
+    /// Record a STORED history sample's ring-time toward the resume cursor. Call ONLY where a sample
+    /// resolved a REAL anchored time (never a wall-clock fallback). Ignores corrupt (over-ceiling) times,
+    /// and flags a reboot when a real sample predates `resumeCursorAtFetchStart` (0 = full pull, no floor).
+    public mutating func noteStoredRingTime(_ rt: UInt32, resumeCursorAtFetchStart: UInt32) {
+        guard rt <= Self.maxPlausibleResumeTicks else { return }
+        if rt > maxStoredRingTime { maxStoredRingTime = rt }
+        if resumeCursorAtFetchStart > 0, rt < resumeCursorAtFetchStart { sawPreResumeData = true }
+    }
+
+    /// The cursor to persist at drain end, given the current cursor and whether `maxStoredRingTime`
+    /// resolves under the CURRENT anchor. A reboot (`sawPreResumeData`) resets to 0 (honest full pull next
+    /// connect); otherwise the cursor advances to `maxStoredRingTime` only if it moved forward AND
+    /// resolves. Unchanged in every other case.
+    public func resumeCursorAtDrainEnd(currentCursor: UInt32, resolvesUnderAnchor: Bool) -> UInt32 {
+        if sawPreResumeData { return 0 }
+        if maxStoredRingTime > currentCursor, resolvesUnderAnchor { return maxStoredRingTime }
+        return currentCursor
+    }
+
+    /// Sanitize a cursor loaded from persistence: a value above the plausibility ceiling is pre-fix
+    /// garbage and must reset to a full pull.
+    public static func sanitizeLoadedCursor(_ persisted: UInt32) -> UInt32 {
+        persisted <= maxPlausibleResumeTicks ? persisted : 0
+    }
+}

Packages/OuraProtocol/Tests/OuraProtocolTests/OuraHistoryDrainTests.swift

@@ -0,0 +1,127 @@
+import XCTest
+@testable import OuraProtocol
+
+/// Pins the Oura history drain + resume-cursor decisions (#91 / #291) that used to live untested inside
+/// OuraLiveSource and silently regressed in a BLE refactor. Pure decisions — no ring, no BLE.
+final class OuraHistoryDrainTests: XCTestCase {
+
+    // MARK: drain continuation
+
+    func testDrainCompletesWhenBytesLeftZero() {
+        var d = OuraHistoryDrain()
+        // moreData == false is the healthy end, regardless of the counters.
+        XCTAssertFalse(d.onSummary(bytesLeft: 0, moreData: false, elapsedSeconds: 0))
+    }
+
+    func testDrainContinuesWhileBytesLeftShrinks() {
+        var d = OuraHistoryDrain()
+        XCTAssertTrue(d.onSummary(bytesLeft: 400_873, moreData: true, elapsedSeconds: 1))
+        XCTAssertTrue(d.onSummary(bytesLeft: 200_000, moreData: true, elapsedSeconds: 2))
+        XCTAssertTrue(d.onSummary(bytesLeft: 1, moreData: true, elapsedSeconds: 3))
+    }
+
+    func testStallGuardStopsAfterMaxFlatSummaries() {
+        var d = OuraHistoryDrain()
+        XCTAssertTrue(d.onSummary(bytesLeft: 1000, moreData: true, elapsedSeconds: 1)) // sets the floor
+        // Same bytes_left repeated — no progress. Stops on the maxStallSummaries-th flat read.
+        XCTAssertTrue(d.onSummary(bytesLeft: 1000, moreData: true, elapsedSeconds: 2))  // stall 1
+        XCTAssertTrue(d.onSummary(bytesLeft: 1000, moreData: true, elapsedSeconds: 3))  // stall 2
+        XCTAssertFalse(d.onSummary(bytesLeft: 1000, moreData: true, elapsedSeconds: 4)) // stall 3 -> stop
+    }
+
+    func testStallCounterResetsOnFreshProgress() {
+        var d = OuraHistoryDrain()
+        XCTAssertTrue(d.onSummary(bytesLeft: 1000, moreData: true, elapsedSeconds: 1))
+        XCTAssertTrue(d.onSummary(bytesLeft: 1000, moreData: true, elapsedSeconds: 2)) // stall 1
+        XCTAssertTrue(d.onSummary(bytesLeft: 900, moreData: true, elapsedSeconds: 3))  // progress -> reset
+        XCTAssertTrue(d.onSummary(bytesLeft: 900, moreData: true, elapsedSeconds: 4))  // stall 1 again
+        XCTAssertTrue(d.onSummary(bytesLeft: 900, moreData: true, elapsedSeconds: 5))  // stall 2 (not stopped)
+    }
+
+    func testDeadlineGuardStopsPastMaxDrainSeconds() {
+        var d = OuraHistoryDrain()
+        XCTAssertTrue(d.onSummary(bytesLeft: 500, moreData: true, elapsedSeconds: 299))
+        XCTAssertFalse(d.onSummary(bytesLeft: 400, moreData: true,
+                                   elapsedSeconds: OuraHistoryDrain.maxDrainSeconds + 0.1))
+    }
+
+    // MARK: stored ring-time → cursor
+
+    func testNoteStoredRingTimeTracksMaxAndIgnoresCorrupt() {
+        var d = OuraHistoryDrain()
+        d.noteStoredRingTime(100, resumeCursorAtFetchStart: 0)
+        d.noteStoredRingTime(3_453_828, resumeCursorAtFetchStart: 0)
+        d.noteStoredRingTime(50, resumeCursorAtFetchStart: 0) // older, doesn't lower the max
+        XCTAssertEqual(d.maxStoredRingTime, 3_453_828)
+        d.noteStoredRingTime(OuraHistoryDrain.maxPlausibleResumeTicks + 1, resumeCursorAtFetchStart: 0)
+        XCTAssertEqual(d.maxStoredRingTime, 3_453_828, "over-ceiling ring-time must be ignored")
+        XCTAssertFalse(d.sawPreResumeData)
+    }
+
+    func testPreResumeDataFlagsReboot() {
+        var d = OuraHistoryDrain()
+        // We sought from cursor 1000; a real sample at 500 is OLDER than the seek → ring reset / seek ignored.
+        d.noteStoredRingTime(500, resumeCursorAtFetchStart: 1000)
+        XCTAssertTrue(d.sawPreResumeData)
+    }
+
+    func testFullPullSeekNeverFlagsReboot() {
+        var d = OuraHistoryDrain()
+        // A full pull seeks from 0 (no floor); early samples must NOT be treated as pre-resume.
+        d.noteStoredRingTime(500, resumeCursorAtFetchStart: 0)
+        XCTAssertFalse(d.sawPreResumeData)
+    }
+
+    // MARK: cursor commit
+
+    func testResumeCursorAdvancesWhenForwardAndResolves() {
+        var d = OuraHistoryDrain()
+        d.noteStoredRingTime(3_453_828, resumeCursorAtFetchStart: 1000)
+        XCTAssertEqual(d.resumeCursorAtDrainEnd(currentCursor: 1000, resolvesUnderAnchor: true), 3_453_828)
+    }
+
+    func testResumeCursorUnchangedWhenNotResolving() {
+        var d = OuraHistoryDrain()
+        d.noteStoredRingTime(3_453_828, resumeCursorAtFetchStart: 1000)
+        XCTAssertEqual(d.resumeCursorAtDrainEnd(currentCursor: 1000, resolvesUnderAnchor: false), 1000)
+    }
+
+    func testResumeCursorUnchangedWhenNotForward() {
+        var d = OuraHistoryDrain()
+        // Newest stored sample EQUALS the cursor (no new data past it) — not forward, and not a reboot
+        // (a sample strictly BELOW the seek would flag `sawPreResumeData`, tested separately).
+        d.noteStoredRingTime(1000, resumeCursorAtFetchStart: 1000)
+        XCTAssertFalse(d.sawPreResumeData)
+        XCTAssertEqual(d.resumeCursorAtDrainEnd(currentCursor: 1000, resolvesUnderAnchor: true), 1000)
+    }
+
+    func testRebootResetsCursorToFullPull() {
+        var d = OuraHistoryDrain()
+        d.noteStoredRingTime(3_453_828, resumeCursorAtFetchStart: 1000) // forward...
+        d.noteStoredRingTime(500, resumeCursorAtFetchStart: 1000)       // ...but also a pre-resume sample
+        XCTAssertTrue(d.sawPreResumeData)
+        XCTAssertEqual(d.resumeCursorAtDrainEnd(currentCursor: 1000, resolvesUnderAnchor: true), 0,
+                       "a reboot forces 0 (full pull) even if a forward sample also arrived")
+    }
+
+    // MARK: loaded-cursor sanitize + reset
+
+    func testSanitizeLoadedCursor() {
+        XCTAssertEqual(OuraHistoryDrain.sanitizeLoadedCursor(3_453_828), 3_453_828)
+        XCTAssertEqual(OuraHistoryDrain.sanitizeLoadedCursor(0), 0)
+        XCTAssertEqual(OuraHistoryDrain.sanitizeLoadedCursor(OuraHistoryDrain.maxPlausibleResumeTicks + 1), 0,
+                       "a persisted cursor above the ceiling is pre-fix garbage → full pull")
+    }
+
+    func testResetClearsState() {
+        var d = OuraHistoryDrain()
+        d.noteStoredRingTime(3_453_828, resumeCursorAtFetchStart: 1000)
+        d.noteStoredRingTime(500, resumeCursorAtFetchStart: 1000)
+        _ = d.onSummary(bytesLeft: 1000, moreData: true, elapsedSeconds: 1)
+        d.reset()
+        XCTAssertEqual(d.maxStoredRingTime, 0)
+        XCTAssertFalse(d.sawPreResumeData)
+        // Stall floor reset: a fresh flat sequence starts counting from scratch.
+        XCTAssertTrue(d.onSummary(bytesLeft: 1000, moreData: true, elapsedSeconds: 1))
+    }
+}