Merge pull request 'Health Connect: Recover late-arriving step samples to fix step count discrepancies' (#6215)

Reviewed-on: https://codeberg.org/Freeyourgadget/Gadgetbridge/pulls/6215

app/src/main/java/nodomain/freeyourgadget/gadgetbridge/service/devices/xiaomi/activity/XiaomiActivityFileFetcher.java

@@ -34,6 +34,7 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.PriorityQueue;
 import java.util.Queue;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import nodomain.freeyourgadget.gadgetbridge.R;
 import nodomain.freeyourgadget.gadgetbridge.impl.GBDevice;
@@ -52,6 +53,8 @@ public class XiaomiActivityFileFetcher {
     private final Queue<XiaomiActivityFileId> mFetchQueue = new PriorityQueue<>();
     private ByteArrayOutputStream mBuffer = new ByteArrayOutputStream();
     private boolean isFetching = false;
+    private volatile boolean awaitingPastResponse = false;
+    private final AtomicBoolean queueHeld = new AtomicBoolean(false);
 
     private final Handler timeoutHandler = new Handler(Looper.getMainLooper());
 
@@ -61,6 +64,8 @@ public class XiaomiActivityFileFetcher {
 
     public void dispose() {
         clearTimeout();
+        awaitingPastResponse = false;
+        queueHeld.set(false);
     }
 
     private void clearTimeout() {
@@ -71,6 +76,7 @@ public class XiaomiActivityFileFetcher {
         // #4305 - Set the timeout in case the watch does not send the file
         this.timeoutHandler.postDelayed(() -> {
             LOG.warn("Timed out waiting for activity file with {} bytes in the buffer", mBuffer.size());
+            awaitingPastResponse = false;
             triggerNextFetch();
         }, 5000L);
     }
@@ -157,6 +163,30 @@ public class XiaomiActivityFileFetcher {
         triggerNextFetch();
     }
 
+    public void setAwaitingPastResponse(final boolean awaiting) {
+        this.awaitingPastResponse = awaiting;
+    }
+
+    public boolean isFetching() {
+        return isFetching;
+    }
+
+    public void signalComplete() {
+        LOG.debug("Nothing more to fetch");
+        isFetching = false;
+        queueHeld.set(false);
+        mHealthService.getSupport().getDevice().unsetBusyTask();
+        GB.signalActivityDataFinish(mHealthService.getSupport().getDevice());
+        GB.updateTransferNotification(null, "", false, 100, mHealthService.getSupport().getContext());
+        mHealthService.getSupport().getDevice().sendDeviceUpdateIntent(mHealthService.getSupport().getContext());
+    }
+
+    public void resumeFetching() {
+        if (queueHeld.compareAndSet(true, false)) {
+            triggerNextFetch();
+        }
+    }
+
     public void fetch(final List<XiaomiActivityFileId> fileIds) {
         // #4305 - ensure unique files
         for (final XiaomiActivityFileId fileId : fileIds) {
@@ -186,12 +216,12 @@ public class XiaomiActivityFileFetcher {
         final XiaomiActivityFileId fileId = mFetchQueue.poll();
 
         if (fileId == null) {
-            LOG.debug("Nothing more to fetch");
-            isFetching = false;
-            mHealthService.getSupport().getDevice().unsetBusyTask();
-            GB.signalActivityDataFinish(mHealthService.getSupport().getDevice());
-            GB.updateTransferNotification(null, "", false, 100, mHealthService.getSupport().getContext());
-            mHealthService.getSupport().getDevice().sendDeviceUpdateIntent(mHealthService.getSupport().getContext());
+            if (awaitingPastResponse) {
+                LOG.debug("Queue empty but awaiting past fetch response, holding signal");
+                queueHeld.set(true);
+                return;
+            }
+            signalComplete();
             return;
         }
 

app/src/main/java/nodomain/freeyourgadget/gadgetbridge/service/devices/xiaomi/services/XiaomiHealthService.java

@@ -869,11 +869,18 @@ public class XiaomiHealthService extends AbstractXiaomiService {
             }
             fileIds.add(fileId);
         }
+        if (subtype == CMD_ACTIVITY_FETCH_TODAY) {
+            activityFetcher.setAwaitingPastResponse(true);
+        }
+
         activityFetcher.fetch(fileIds);
 
         if (subtype == CMD_ACTIVITY_FETCH_TODAY) {
             LOG.debug("Fetch recorded data from the past");
             fetchRecordedDataPast();
+        } else if (subtype == CMD_ACTIVITY_FETCH_PAST) {
+            activityFetcher.setAwaitingPastResponse(false);
+            activityFetcher.resumeFetching();
         }
     }
 

app/src/main/java/nodomain/freeyourgadget/gadgetbridge/util/healthconnect/HealthConnectUtils.kt

@@ -50,6 +50,7 @@ import java.time.Instant
 import java.time.ZoneId
 import java.time.ZonedDateTime
 import java.time.format.DateTimeFormatter
+import java.time.temporal.ChronoUnit
 import java.util.*
 import java.util.function.BiConsumer
 import kotlin.math.pow
@@ -410,6 +411,7 @@ class HealthConnectUtils {
                         val sliceTotalSynced = sliceStats.sumOf { it.recordsSynced }
 
                         val latestRecordTs = sliceStats.mapNotNull { it.latestRecordTimestamp }.maxOrNull()
+
                         if (latestRecordTs != null && latestRecordTs.isAfter(timestampToPersistForThisDataType)) {
                             timestampToPersistForThisDataType = latestRecordTs
                         } else if (sliceTotalSynced == 0) {

app/src/main/java/nodomain/freeyourgadget/gadgetbridge/util/healthconnect/syncers/AbstractActivitySampleSyncer.kt

@@ -24,6 +24,7 @@ import nodomain.freeyourgadget.gadgetbridge.impl.GBDevice
 import nodomain.freeyourgadget.gadgetbridge.model.ActivitySample
 import nodomain.freeyourgadget.gadgetbridge.util.healthconnect.HealthConnectUtils
 import org.slf4j.Logger
+import java.time.Duration
 import java.time.Instant
 import java.time.ZoneId
 import java.time.ZoneOffset
@@ -34,11 +35,29 @@ internal abstract class AbstractActivitySampleSyncer<TRecord : Record> : Activit
     protected abstract val logger: Logger
     protected abstract val recordClass: KClass<TRecord>
 
+    // Re-emit samples this far before the slice start. The ACTIVITY cursor is shared across
+    // steps/calories/distance and advances to the furthest delivered; when step or distance detail
+    // trails the calorie summary, those minutes would be clipped and lost. The per-minute
+    // clientRecordId makes the re-emitted overlap an upsert. Heart rate is a series keyed on its
+    // start time, does not extend this base, and keeps strict boundaries.
+    protected open val lateSampleLookback: Duration = Duration.ofHours(1)
+
+    internal fun isWithinSlice(
+        endTs: Instant,
+        startTs: Instant,
+        sliceStartBoundary: Instant,
+        sliceEndBoundary: Instant
+    ): Boolean {
+        val effectiveStart = sliceStartBoundary.minus(lateSampleLookback)
+        return !endTs.isBefore(effectiveStart) && !startTs.isAfter(sliceEndBoundary)
+    }
+
     internal abstract fun convertSample(
         sample: ActivitySample,
         offset: ZoneOffset,
         metadata: Metadata,
-        deviceName: String
+        deviceName: String,
+        version: Long
     ): TRecord?
 
     override suspend fun sync(
@@ -69,6 +88,13 @@ internal abstract class AbstractActivitySampleSyncer<TRecord : Record> : Activit
 
         logger.info("Processing ${relevantSamples.size} samples for steps for device '$deviceName' for slice $sliceStartBoundary to $sliceEndBoundary.")
 
+        // clientRecordVersion: every record in this slice shares the run's wall-clock so a later
+        // sync always outranks the value it previously wrote for a minute. HC keeps the highest
+        // version on a clientRecordId collision, so the freshest (most complete) re-read wins even
+        // when a value is revised downward. Must not be the metric value itself: a downward
+        // correction would then carry a lower version and be silently ignored.
+        val recordVersion = System.currentTimeMillis()
+
         val recordsToInsert = mutableListOf<Record>()
         var skippedCount = 0
         var latestSyncedTimestamp: Instant? = null
@@ -76,7 +102,7 @@ internal abstract class AbstractActivitySampleSyncer<TRecord : Record> : Activit
             val endTs = Instant.ofEpochSecond(currentSample.timestamp.toLong())
             val startTs = endTs.minus(1, ChronoUnit.MINUTES)
 
-            if (endTs.isBefore(sliceStartBoundary) || startTs.isAfter(sliceEndBoundary)) {
+            if (!isWithinSlice(endTs, startTs, sliceStartBoundary, sliceEndBoundary)) {
                 logger.trace(
                     "Skipping {} for device '{}' for sample at {} (interval {} to {}) as its interval is outside the slice {} - {}.",
                     recordTypeName,
@@ -90,7 +116,7 @@ internal abstract class AbstractActivitySampleSyncer<TRecord : Record> : Activit
                 return@forEach
             }
 
-            val record = convertSample(sample = currentSample, offset.rules.getOffset(endTs), metadata, deviceName)
+            val record = convertSample(sample = currentSample, offset.rules.getOffset(endTs), metadata, deviceName, recordVersion)
             if (record == null) {
                 skippedCount++
                 return@forEach

app/src/main/java/nodomain/freeyourgadget/gadgetbridge/util/healthconnect/syncers/ActiveCaloriesSyncer.kt

@@ -35,7 +35,8 @@ internal object ActiveCaloriesSyncer : AbstractActivitySampleSyncer<ActiveCalori
         sample: ActivitySample,
         offset: ZoneOffset,
         metadata: Metadata,
-        deviceName: String
+        deviceName: String,
+        version: Long
     ): ActiveCaloriesBurnedRecord? {
         val caloriesInMinute = sample.activeCalories
         if (caloriesInMinute <= 0) {
@@ -56,7 +57,7 @@ internal object ActiveCaloriesSyncer : AbstractActivitySampleSyncer<ActiveCalori
             endTs,
             offset,
             Energy.calories(caloriesInMinute.toDouble()),
-            metadata
+            clientRecordMetadata(metadata, "calories", endTs.epochSecond, version)
         )
     }
 }

app/src/main/java/nodomain/freeyourgadget/gadgetbridge/util/healthconnect/syncers/DistanceSyncer.kt

@@ -35,7 +35,8 @@ internal object DistanceSyncer : AbstractActivitySampleSyncer<DistanceRecord>()
         sample: ActivitySample,
         offset: ZoneOffset,
         metadata: Metadata,
-        deviceName: String
+        deviceName: String,
+        version: Long
     ): DistanceRecord? {
         val distanceCm = sample.distanceCm
         if (distanceCm <= 0 || distanceCm == ActivitySample.NOT_MEASURED) {
@@ -56,7 +57,7 @@ internal object DistanceSyncer : AbstractActivitySampleSyncer<DistanceRecord>()
             endTime = endTs,
             endZoneOffset = offset,
             distance = Length.meters(distanceCm / 100.0),
-            metadata = metadata
+            metadata = clientRecordMetadata(metadata, "distance", endTs.epochSecond, version)
         )
     }
 }

app/src/main/java/nodomain/freeyourgadget/gadgetbridge/util/healthconnect/syncers/HealthConnectSyncer.kt

@@ -24,6 +24,25 @@ import nodomain.freeyourgadget.gadgetbridge.model.ActivitySample
 import java.time.Instant
 import java.time.ZoneId
 
+// Deterministic clientRecordId so re-emitting an overlapping window upserts instead of duplicating.
+// version is the clientRecordVersion; HC keeps the highest on conflict (newVersion >= existing
+// overwrites). Callers pass the sync run's wall-clock so a later run always outranks the value it
+// previously wrote for a minute. It must never be the metric value: a downward correction would
+// then carry a lower version and be silently ignored, freezing the minute at its stale maximum.
+internal fun clientRecordMetadata(
+    base: Metadata,
+    type: String,
+    idKey: Long,
+    version: Long
+): Metadata {
+    val device = base.device ?: return base
+    val id = "gb-$type-${device.manufacturer ?: "unknown"}-${device.model ?: "unknown"}-$idKey"
+    return when (base.recordingMethod) {
+        Metadata.RECORDING_METHOD_ACTIVELY_RECORDED -> Metadata.activelyRecorded(device, id, version)
+        else -> Metadata.autoRecorded(device, id, version)
+    }
+}
+
 /**
  * Statistics returned by a syncer after processing a slice.
  */

app/src/main/java/nodomain/freeyourgadget/gadgetbridge/util/healthconnect/syncers/StepsSyncer.kt

@@ -34,7 +34,8 @@ internal object StepsSyncer : AbstractActivitySampleSyncer<StepsRecord>() {
         sample: ActivitySample,
         offset: ZoneOffset,
         metadata: Metadata,
-        deviceName: String
+        deviceName: String,
+        version: Long
     ): StepsRecord? {
         val stepsInMinute = sample.steps.toLong()
         // <= 0 means "no steps in that minute" - common, drop silently.
@@ -55,7 +56,7 @@ internal object StepsSyncer : AbstractActivitySampleSyncer<StepsRecord>() {
             endTs,
             offset,
             stepsInMinute,
-            metadata
+            clientRecordMetadata(metadata, "steps", endTs.epochSecond, version)
         )
     }
 }

app/src/test/java/nodomain/freeyourgadget/gadgetbridge/util/healthconnect/syncers/ActivitySampleSliceBoundaryTest.kt

@@ -0,0 +1,99 @@
+package nodomain.freeyourgadget.gadgetbridge.util.healthconnect.syncers
+
+import androidx.health.connect.client.records.StepsRecord
+import androidx.health.connect.client.records.metadata.Device
+import androidx.health.connect.client.records.metadata.Metadata
+import nodomain.freeyourgadget.gadgetbridge.model.ActivitySample
+import org.junit.Assert.*
+import org.junit.Test
+import java.time.Instant
+import java.time.ZoneOffset
+import kotlin.reflect.KClass
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+class ActivitySampleSliceBoundaryTest {
+
+    private val sliceStart: Instant = Instant.ofEpochSecond(1_700_000_000L)
+    private val sliceEnd: Instant = sliceStart.plusSeconds(14 * 60)
+
+    // Concrete syncer with the default 1h lookback (mirrors Steps/Calories/Distance).
+    private object IdempotentSyncer : AbstractActivitySampleSyncer<StepsRecord>() {
+        override val logger: Logger = LoggerFactory.getLogger("test")
+        override val recordClass: KClass<StepsRecord> = StepsRecord::class
+        override fun convertSample(sample: ActivitySample, offset: ZoneOffset, metadata: Metadata, deviceName: String, version: Long): StepsRecord? = null
+    }
+
+    // Syncer that opts out of the lookback (mirrors a non-idempotent record type).
+    private object StrictSyncer : AbstractActivitySampleSyncer<StepsRecord>() {
+        override val logger: Logger = LoggerFactory.getLogger("test")
+        override val recordClass: KClass<StepsRecord> = StepsRecord::class
+        override val lateSampleLookback = java.time.Duration.ZERO
+        override fun convertSample(sample: ActivitySample, offset: ZoneOffset, metadata: Metadata, deviceName: String, version: Long): StepsRecord? = null
+    }
+
+    private fun within(syncer: AbstractActivitySampleSyncer<*>, endTs: Instant): Boolean =
+        syncer.isWithinSlice(endTs, endTs.minusSeconds(60), sliceStart, sliceEnd)
+
+    @Test
+    fun sampleInsideSlice_emitted() {
+        assertTrue(within(IdempotentSyncer, sliceStart.plusSeconds(300)))
+    }
+
+    @Test
+    fun lateSampleWithinLookback_recoveredByIdempotentSyncer() {
+        // 30 min before the slice start: clipped without lookback, recovered with it.
+        val lateTs = sliceStart.minusSeconds(30 * 60)
+        assertTrue(within(IdempotentSyncer, lateTs))
+    }
+
+    @Test
+    fun lateSampleWithinLookback_clippedByStrictSyncer() {
+        val lateTs = sliceStart.minusSeconds(30 * 60)
+        assertFalse(within(StrictSyncer, lateTs))
+    }
+
+    @Test
+    fun sampleOlderThanLookback_clipped() {
+        // 90 min before start is beyond the 1h window.
+        val tooOld = sliceStart.minusSeconds(90 * 60)
+        assertFalse(within(IdempotentSyncer, tooOld))
+    }
+
+    @Test
+    fun sampleAfterSliceEnd_clippedRegardlessOfLookback() {
+        // Upper bound is never relaxed: a sample whose interval starts after slice end is dropped.
+        val afterEnd = sliceEnd.plusSeconds(120)
+        assertFalse(within(IdempotentSyncer, afterEnd))
+    }
+
+    @Test
+    fun sampleStraddlingSliceEnd_emitted() {
+        // endTs just past sliceEnd but startTs (endTs-60s) still within: kept.
+        val straddle = sliceEnd.plusSeconds(30)
+        assertTrue(within(IdempotentSyncer, straddle))
+    }
+
+    // The clientRecordVersion must carry the supplied version verbatim (the sync run's wall-clock),
+    // never the metric value: HC keeps the highest version on a clientRecordId collision, so a value
+    // revised downward would be silently ignored if the value doubled as the version.
+    @Test
+    fun clientRecordVersion_carriesSuppliedVersion_notMetricValue() {
+        val base = Metadata.autoRecorded(Device(Device.TYPE_WATCH, "Acme", "Band"))
+        val version = 1_700_000_500_000L
+        val meta = clientRecordMetadata(base, "steps", endEpoch, version)
+        assertEquals(version, meta.clientRecordVersion)
+    }
+
+    // The dedup key (clientRecordId) is fixed by type + device + timestamp and must not vary with the
+    // version, so a re-emit of the same minute collides and upserts instead of duplicating.
+    @Test
+    fun clientRecordId_independentOfVersion() {
+        val base = Metadata.autoRecorded(Device(Device.TYPE_WATCH, "Acme", "Band"))
+        val low = clientRecordMetadata(base, "steps", endEpoch, 1L)
+        val high = clientRecordMetadata(base, "steps", endEpoch, 999L)
+        assertEquals(low.clientRecordId, high.clientRecordId)
+    }
+
+    private val endEpoch = sliceStart.epochSecond
+}