Erase ring history after sync and fix per-ring sync stalls

PulseLoop/Events/PulseEventBus.swift

@@ -15,6 +15,9 @@ enum PulseEvent: Sendable {
     case batteryLevel(percent: Int)
     case decodedPacket(RingDecodedEvent)
     case rwfitSyncOutcome(RWfitSyncOutcome)
+    /// Durable-commit barrier: the persistence subscriber saves everything published before it and
+    /// resolves `RingHistoryCommit` with the outcome. Gates erasing history from the ring.
+    case historyCommitBarrier(UUID)
     case rwfitInitialization(RWfitInitializationState)
     case rwfitDiagnostic(message: String, metadata: [String: String])
     case rwfitMeasurementOutcome(RWfitMeasurementOutcome)
@@ -246,10 +249,29 @@ final class EventPersistenceSubscriber {
     }
 
     func persist(_ event: PulseEvent) {
+        if case let .historyCommitBarrier(id) = event {
+            commitBarrier(id)
+            return
+        }
         applyPersist(event)
         scheduleFlush()
     }
 
+    /// Every event published before the barrier has been applied by now (the stream is ordered), so a
+    /// successful throwing save here proves that history is durable and may be erased from the ring.
+    func commitBarrier(_ id: UUID) {
+        flushTask?.cancel()
+        flushTask = nil
+        do {
+            if context.hasChanges { try context.save() }
+            if pendingWrites > 0 { PulseDataChange.shared.notify() }
+            pendingWrites = 0
+            RingHistoryCommit.resolve(id, committed: true)
+        } catch {
+            RingHistoryCommit.resolve(id, committed: false)
+        }
+    }
+
     /// Debounced batch flush: reset the idle timer each event; flush immediately if too many writes
     /// have piled up. Coalesces a sync burst into a handful of saves + change signals.
     private func scheduleFlush() {
@@ -443,6 +465,8 @@ final class EventPersistenceSubscriber {
             }
             seenHistoryKeys.removeAll(keepingCapacity: true)
             DailyCalorieEstimator.flushDirty(context: context)
+        case .historyCommitBarrier:
+            break   // handled in `persist` before `applyPersist` (it must save, not batch)
         case .decodedPacket, .rwfitInitialization, .rwfitDiagnostic, .rwfitMeasurement, .rwfitMeasurementOutcome:
             break
         // `.wearState` is a live condition the measurement flow reacts to, not data — nothing to store.

PulseLoop/RingProtocol/CRPSyncEngine.swift

@@ -43,6 +43,29 @@ final class CRPSyncEngine: RingSyncEngine {
     private var historySyncInFlight = false
     private var historyCompletionTask: Task<Void, Never>?
 
+    /// How long a timing vital gets to return its *first* frame before it stops holding the pass
+    /// open. A vital that never answers (stress `2/47` on zaggash's R11 and on the issue-#58 R100,
+    /// which ignore its state read-back `2/45` too) used to keep every pass waiting out the whole
+    /// `historyWatchdogNanos`. A vital that does answer within the window stays pending until its
+    /// terminal frame, bounded only by that watchdog, so this drops no data the ring offers.
+    ///
+    /// Why a timeout and not the connect read-backs (`2/6` HR, `2/7` HRV, `2/8` SpO2, `2/45` stress,
+    /// `2/21` temperature): the vendor ring SDK (`com.crrepa.ble`) gates nothing on them. Its history
+    /// calls are fire-and-forget writes (`k/f` paces on write completion, not on a reply), each
+    /// parser walks its own frames (`h/c.d`: `index + 1` until the terminal one), and there is no
+    /// "pass complete" at all; the Android port likewise only advances the cursor. The read-backs
+    /// can't stand in for the timeout either: the ring that leaves `2/47` silent leaves `2/45`
+    /// silent as well, a monitor that is on can still go unanswered, and they describe the state
+    /// *before* `applyTimingSettings` re-enables everything.
+    ///
+    /// Five seconds covers the connect handshake queued ahead of the history queries on the single
+    /// write channel plus a stored-frame reply; the ring answers the frame-0 queries in send order,
+    /// ahead of any follow-up frame.
+    private let timingFirstFrameWindowNanos: UInt64
+    /// Timing vitals that have not returned any frame yet in the current pass.
+    private var awaitingFirstTimingFrame: Set<Int> = []
+    private var timingFirstFrameTask: Task<Void, Never>?
+
     /// User-chosen all-day measurement config. Applied in the connect handshake and updatable
     /// live via `applyMeasurementSettings`. `nil` ⇒ the user has never saved one, and a fresh R11
     /// ships with every all-day monitor OFF, so it records nothing to sync. We therefore fall back to
@@ -75,12 +98,14 @@ final class CRPSyncEngine: RingSyncEngine {
         writer: RingCommandWriter?,
         // Below RingSyncCoordinator's 20-second stall deadline so this engine's explicit `done` wins.
         historyWatchdogNanos: UInt64 = 15_000_000_000,
+        timingFirstFrameWindowNanos: UInt64 = 5_000_000_000,
         eventSink: @escaping @MainActor (PulseEvent) -> Void = { event in
             Task { PulseEventBus.shared.publish(event) }
         }
     ) {
         self.writer = writer
         self.historyWatchdogNanos = historyWatchdogNanos
+        self.timingFirstFrameWindowNanos = timingFirstFrameWindowNanos
         self.eventSink = eventSink
     }
 
@@ -184,9 +209,14 @@ final class CRPSyncEngine: RingSyncEngine {
         guard !historySyncInFlight else { return }
         historySyncInFlight = true
         pendingTimingHistoryCommands = Self.timingHistoryCommands
+        awaitingFirstTimingFrame = Self.timingHistoryCommands
         eventSink(.syncProgress(stage: "Syncing ring history…"))
         queryAllHistory(includeSleepBackfill: includeSleepBackfill)
-        armHistoryCompletionWatchdog()
+        // Unsupported/disabled streams are allowed to stay silent. A vital silent past the
+        // first-frame window stops holding the pass; the watchdog bounds one that answers and stalls.
+        // Either way the pass completes with whatever the ring did return instead of wedging sync.
+        timingFirstFrameTask = deadline(after: timingFirstFrameWindowNanos) { $0.dropSilentTimingStreams() }
+        historyCompletionTask = deadline(after: historyWatchdogNanos) { $0.finishHistorySync() }
     }
 
     /// CRP history is a standalone query set, so it participates in the coordinator's periodic and
@@ -195,33 +225,44 @@ final class CRPSyncEngine: RingSyncEngine {
         startHistorySync(includeSleepBackfill: false)
     }
 
-    private func armHistoryCompletionWatchdog() {
-        historyCompletionTask?.cancel()
-        let delay = historyWatchdogNanos
-        historyCompletionTask = Task { @MainActor [weak self] in
-            try? await Task.sleep(nanoseconds: delay)
+    private func deadline(
+        after nanos: UInt64,
+        _ action: @escaping @MainActor (CRPSyncEngine) -> Void
+    ) -> Task<Void, Never> {
+        Task { @MainActor [weak self] in
+            try? await Task.sleep(nanoseconds: nanos)
             guard let self, !Task.isCancelled else { return }
-            // Unsupported/disabled streams are allowed to stay silent. Reaching the bounded deadline
-            // completes the poll with whatever the ring did return instead of wedging sync forever.
-            self.finishHistorySync()
+            action(self)
+        }
+    }
+
+    /// Stop waiting on every timing vital that hasn't returned a single frame by the end of the
+    /// first-frame window (see `timingFirstFrameWindowNanos`). A late reply from a dropped vital is
+    /// still decoded and persisted by the bridge, and `handle` still walks it while the pass is open.
+    private func dropSilentTimingStreams() {
+        guard historySyncInFlight else { return }
+        pendingTimingHistoryCommands.subtract(awaitingFirstTimingFrame)
+        awaitingFirstTimingFrame.removeAll()
+        if pendingTimingHistoryCommands.isEmpty {
+            finishHistorySync()
         }
     }
 
     private func finishHistorySync() {
         guard historySyncInFlight else { return }
-        historySyncInFlight = false
-        pendingTimingHistoryCommands.removeAll()
+        cancelHistorySync()
         requestedTimingFrames.removeAll()
-        historyCompletionTask?.cancel()
-        historyCompletionTask = nil
         eventSink(.syncProgress(stage: "done"))
     }
 
     private func cancelHistorySync() {
         historySyncInFlight = false
         pendingTimingHistoryCommands.removeAll()
+        awaitingFirstTimingFrame.removeAll()
         historyCompletionTask?.cancel()
         historyCompletionTask = nil
+        timingFirstFrameTask?.cancel()
+        timingFirstFrameTask = nil
     }
 
     /// Whether older nights have already been backfilled on this GATT link.
@@ -283,6 +324,8 @@ final class CRPSyncEngine: RingSyncEngine {
         guard historySyncInFlight,
               case let .timingHistoryFrame(cmd, day, frameIndex) = event,
               Self.timingHistoryCommands.contains(cmd) else { return }
+        // Any frame proves the vital answers, so the first-frame window no longer applies to it.
+        awaitingFirstTimingFrame.remove(cmd)
         if day == CRPCommands.historyDayToday,
            frameIndex >= terminalFrameIndex(cmd: cmd) {
             pendingTimingHistoryCommands.remove(cmd)

PulseLoop/RingProtocol/ColmiDecoder.swift

@@ -248,6 +248,10 @@ struct ColmiDecoder {
             return decodeSpo2(v, calendar: calendar)
         case ColmiCommandID.bigDataSleep:
             return decodeSleep(v, calendar: calendar)
+        case ColmiCommandID.bigDataSleepNap:
+            // QRing's `parseDaySleepLunch` reads the same `[daysAgo][len][start][end][pairs…]` record
+            // as the main-sleep action (PulseLoopAndroid `ColmiDecoder.kt:337-341`).
+            return decodeSleep(v, calendar: calendar)
         case ColmiCommandID.bigDataTemperature:
             return decodeTemperature(v, calendar: calendar)
         default:

PulseLoop/RingProtocol/ColmiEncoder.swift

@@ -35,7 +35,7 @@
     }
 
     /// User profile. Sensible neutral defaults; PulseLoop can wire real profile values later.
     func userPreferences(metric: Bool = true, gender: UInt8 = 0x02, age: UInt8 = 25, heightCm: UInt8 = 175, weightKg: UInt8 = 70) -> [UInt8] {
         [
             ColmiCommandID.preferences,
             ColmiCommandID.prefWrite,
@@ -137,13 +137,49 @@
 
     // MARK: Big-data requests (raw, sent on the command characteristic)
 
+    /// `bc 2a 01 00 ff 00 ff` — all-history SpO2 (payload `ff`).
     func bigDataSpo2() -> Data {
-        Data([ColmiCommandID.bigDataV2, ColmiCommandID.bigDataSpo2, 0x01, 0x00, 0xff, 0x00, 0xff])
+        Self.bigDataRequest(ColmiCommandID.bigDataSpo2, payload: [0xff])
     }
+
+    /// `bc 27 02 00 81 80 ff 01` — QRing's New_Sleep_Protocol request, `LargeDataHandler.addHeader(39,
+    /// [0xff, 0x01])`: `ff` = all history, `01` = protocol version (PulseLoopAndroid
+    /// `ColmiEncoder.kt:261-268`, ported from the QRing decompile; its `ColmiSleepSyncTest.kt:47-58`
+    /// pins the same bytes). The 1-byte `bc 27 01 00 ff 00 ff` we used to send (GadgetBridge
+    /// `YawellRingDeviceSupport.fetchHistorySleep`) is silently ignored by newer firmware, so sleep
+    /// never synced there. The reply is the same `0x27` day-record frame `decodeSleep` reads
+    /// (`testRing2ProCapturedNightDecodes` is a reply to this request), plus a `0x3e` nap frame on
+    /// rings that file short sleeps separately.
     func bigDataSleep() -> Data {
-        Data([ColmiCommandID.bigDataV2, ColmiCommandID.bigDataSleep, 0x01, 0x00, 0xff, 0x00, 0xff])
+        Self.bigDataRequest(ColmiCommandID.bigDataSleep, payload: [0xff, 0x01])
     }
+
+    /// `bc 25 01 00 3e 81 02` — legacy temperature series (payload = day count 2).
     func bigDataTemperature() -> Data {
-        Data([ColmiCommandID.bigDataV2, ColmiCommandID.bigDataTemperature, 0x01, 0x00, 0x3e, 0x81, 0x02])
+        Self.bigDataRequest(ColmiCommandID.bigDataTemperature, payload: [0x02])
+    }
+
+    /// `bc <type> <len u16 LE> <crc u16 LE> <payload>` — QRing's `LargeDataHandler.addHeader`, where
+    /// the CRC is CRC16/MODBUS over the payload (PulseLoopAndroid `ColmiEncoder.kt:237-256`).
+    static func bigDataRequest(_ type: UInt8, payload: [UInt8]) -> Data {
+        let crc = crc16Modbus(payload)
+        let len = payload.count
+        return Data([
+            ColmiCommandID.bigDataV2, type,
+            UInt8(len & 0xff), UInt8((len >> 8) & 0xff),
+            UInt8(crc & 0xff), UInt8((crc >> 8) & 0xff),
+        ] + payload)
+    }
+
+    /// CRC16/MODBUS: init `0xffff`, reflected poly `0xa001`, no final xor.
+    static func crc16Modbus(_ bytes: [UInt8]) -> UInt16 {
+        var crc: UInt16 = 0xffff
+        for byte in bytes {
+            crc ^= UInt16(byte)
+            for _ in 0..<8 {
+                crc = crc & 1 != 0 ? (crc >> 1) ^ 0xa001 : crc >> 1
+            }
+        }
+        return crc
     }
 }

PulseLoop/RingProtocol/ColmiProtocol.swift

@@ -64,6 +64,9 @@ enum ColmiCommandID {
     // big-data types
     static let bigDataTemperature: UInt8 = 0x25
     static let bigDataSleep: UInt8 = 0x27
+    /// Nap ("lunch") sleep, QRing big-data action 62. Rings with nap support file sessions of
+    /// 180 minutes or less here, unsolicited alongside the `0x27` reply; same day-record layout.
+    static let bigDataSleepNap: UInt8 = 0x3e
     static let bigDataSpo2: UInt8 = 0x2a
 
     // `0x69`/`0x6A` reading types for on-demand `SpotVital`s (QRing `StartHeartRateReq.getSimpleReq`).

PulseLoop/RingProtocol/ColmiSmartHealthCoordinator.swift

@@ -137,7 +137,8 @@ final class ColmiSmartHealthCoordinator: WearableCoordinator {
     ///   R99 proved this benign: it NAKed `01 45` (HRV) and `01 1c` (all-day BP) with `0xFC` and honoured
     ///   the other three.
     /// - `.spo2History` is the all-day `05 1A` log. A ring without it answers the query with a no-data
-    ///   header or `0xFC`, which `YCBTHistoryTransfer` skips permanently.
+    ///   header or `0xFC`, which `YCBTHistoryTransfer` skips permanently — or, like the R99, with nothing
+    ///   at all, which costs one watchdog and is then not asked again until the link drops.
     ///
     /// Neither is named by any bit in `YCBTSupportFunction`, so gating them would not defer the decision
     /// — it would make them permanently unreachable (see `bitmapGatedCapabilities`).

PulseLoop/RingProtocol/ColmiSyncEngine.swift

@@ -54,6 +54,9 @@
     private var daysAgo = 0
     /// The midnight of the day currently being synced (for HR/stress/HRV sample timestamps).
     private var syncDay = Calendar.current.startOfDay(for: Date())
+    /// How many packets the current HR/stress/HRV day has, from its packet 0 (`v[2]`); nil until that
+    /// header arrives. Reset on every day request. Drives `isTerminalPacket`.
+    private var expectedPackets: Int?
     /// A targeted post-workout run: today's HR log → SpO2 big-data → done, skipping the other five
     /// stages. Set by `startVitalsBackfill()`, cleared whenever a run finishes or a full sync starts.
     private var isVitalsBackfill = false
@@ -192,12 +195,12 @@
         case .activity:
             daysAgo = 0; stage = .heartRate; requestHeartRate()
         case .heartRate:
             if isVitalsBackfill { stage = .spo2; requestSpo2() }
             else { stage = .stress; requestStress() }
         case .stress:
             stage = .spo2; requestSpo2()
         case .spo2:
             if isVitalsBackfill { finishSync() }
             else { stage = .sleep; requestSleep() }
         case .sleep:
             daysAgo = 0; stage = .hrv; requestHRV()
@@ -219,22 +222,35 @@
 
     private func requestActivity() {
         syncDay = dayStart(daysAgo)
+        expectedPackets = nil
         writer?.enqueue(Data(encoder.syncActivity(daysAgo: daysAgo)))
     }
 
     private func requestHeartRate() {
         syncDay = dayStart(daysAgo)
-        let unix = Int(syncDay.timeIntervalSince1970)
+        expectedPackets = nil
+        let unix = Self.heartRateRequestEpoch(dayStart: syncDay, timeZone: calendar.timeZone)
         writer?.enqueue(Data(encoder.syncHeartRate(fromUnix: unix)))
     }
 
+    /// The ring indexes its daily HR logs by local wall-clock time read as UTC, so the request carries
+    /// local midnight *plus* the zone offset — the calendar date at 00:00 UTC. GadgetBridge
+    /// (`YawellRingDeviceSupport.java:916`, `+ ZONE_OFFSET + DST_OFFSET`) and PulseLoopAndroid
+    /// (`ColmiSyncEngine.kt:496-500`, `atStartOfDay(ZoneOffset.UTC)`, from QRing) agree. The true
+    /// local-midnight epoch we sent before falls inside the ring's *previous* day east of UTC.
+    static func heartRateRequestEpoch(dayStart: Date, timeZone: TimeZone) -> Int {
+        Int(dayStart.timeIntervalSince1970) + timeZone.secondsFromGMT(for: dayStart)
+    }
+
     private func requestStress() {
         syncDay = calendar.startOfDay(for: Date())
+        expectedPackets = nil
         writer?.enqueue(Data(encoder.syncStress()))
     }
 
     private func requestHRV() {
         syncDay = dayStart(daysAgo)
+        expectedPackets = nil
         writer?.enqueue(Data(encoder.syncHRV(daysAgo: daysAgo)))
     }
 
@@ -251,6 +267,7 @@
     /// Decode a paged history frame for the current sync-day and advance the machine. Returns the
     /// decoded sample events for the bus.
     func handleHistoryFrame(_ data: Data) -> [RingDecodedEvent] {
+        captureExpectedPackets(data)
         let events = decoder.decodeHistory(data, day: syncDay, calendar: calendar)
         advanceAfterPagedFrame(data)
         armWatchdog()   // progress made — reset the stall timer
@@ -284,48 +301,71 @@
         lastRealtimeFrameAt = Date()
     }
 
+    /// HR/stress/HRV packet 0 is a header: `<op> 00 <size> <cadence min> …`, where `size` counts the
+    /// day's packets *including* this header (QRing `ReadHeartRateRsp`/`PressureRsp`/`HRVRsp` per
+    /// PulseLoopAndroid `ColmiSyncEngine.kt:357-372`; GadgetBridge reads the same `v[2]` for HR at
+    /// `YawellRingDeviceSupport.java:232` and HRV at `YawellRingPacketHandler.java:541`; tahnok
+    /// `hr.py` too). Out-of-range sizes are ignored so a garbled header falls back to the defaults.
+    private func captureExpectedPackets(_ data: Data) {
+        guard let packet = ColmiPacket(validating: data) else { return }
+        let v = packet.bytes
+        switch v[0] {
+        case ColmiCommandID.syncHeartRate, ColmiCommandID.syncStress, ColmiCommandID.syncHRV:
+            guard v[1] == 0 else { return }
+            let size = Int(v[2])
+            expectedPackets = (2...64).contains(size) ? size : nil
+        default:
+            return
+        }
+    }
+
     /// Paged stages (activity/HR/stress/HRV) advance day-by-day, then to the next stage. We treat an
-    /// empty marker (0xff) OR the final packet as "this day done".
+    /// empty marker (0xff) OR the final packet as "this day done". A header packet (0) just waits for
+    /// data packets.
     private func advanceAfterPagedFrame(_ data: Data) {
         let packetNr = ColmiDecoder.historyPacketNumber(data)
         let isEmpty = packetNr == 0xff
-        // For paged HR/stress/HRV, byte[2] in packet 0 carries the total; we conservatively treat any
-        // empty/terminal as "advance". A header packet (0) just waits for data packets.
         let dayComplete = isEmpty || isTerminalPacket(data)
 
         guard dayComplete else { return }
 
         switch stage {
         case .activity:
             if daysAgo < 7 { daysAgo += 1; requestActivity() }
             else { daysAgo = 0; stage = .heartRate; requestHeartRate() }
         case .heartRate:
             if isVitalsBackfill { stage = .spo2; requestSpo2() }
             else if daysAgo < 7 { daysAgo += 1; requestHeartRate() }
             else { stage = .stress; requestStress() }
         case .stress:
             stage = .spo2; requestSpo2()
         case .hrv:
             if daysAgo < 6 { daysAgo += 1; requestHRV() }
             else { stage = .temperature; requestTemperature() }
         default:
             break
         }
     }
 
-    /// Terminal packet heuristic for paged stages: stress/HRV finish at packet 4 (GadgetBridge);
-    /// activity finishes when `currentPacket == totalPackets - 1` (bytes 5/6).
+    /// A day's last data packet: `packetNr == size - 1`, with `size` from the day's header
+    /// (`captureExpectedPackets`) — QRing's rule (PulseLoopAndroid `ColmiSyncEngine.kt:565-588`),
+    /// GadgetBridge's `hrPacketNr == packetsTotalNr - 1` (`YawellRingDeviceSupport.java:290`) and
+    /// tahnok's `sub_type == self.size - 1`. HR used to have no terminal at all, so every HR day with
+    /// data sat out the 10 s watchdog, which then skipped the whole stage and days 1-7 were never
+    /// requested. Fallbacks when the header was missed: stress/HRV keep the old size 5 (48 half-hour
+    /// slots = 12 + 3 × 13 values); HR ends on packet 23 (a full 5-minute day is 9 + 22 × 13 ≥ 288
+    /// values), else the watchdog. Activity ends when `currentPacket == totalPackets - 1` (bytes 5/6).
     private func isTerminalPacket(_ data: Data) -> Bool {
         let v = [UInt8](data)
         guard v.count >= 7 else { return false }
+        let packetNr = Int(v[1])
         switch v[0] {
         case ColmiCommandID.syncStress, ColmiCommandID.syncHRV:
-            return Int(v[1]) == 4
+            return packetNr == (expectedPackets ?? 5) - 1
         case ColmiCommandID.syncActivity:
             return Int(v[5]) == Int(v[6]) - 1
         case ColmiCommandID.syncHeartRate:
-            // packet 0 carries total in byte[2]; treat last packet as terminal.
-            return false   // UNVERIFIED: rely on emptiness / next-day request cadence
+            return packetNr == (expectedPackets ?? 24) - 1
         default:
             return false
         }