Fix #5797: Advance HC sync cursor by last-written record timestamp

The sync state cursor was advancing to the slice end timestamp
unconditionally, even when no records were written. Late-arriving data
(sleep, workouts, VO2Max, etc.) that arrives after the cursor has
already advanced past it would be permanently skipped.
Each syncer now reports the latest timestamp of records actually
inserted into Health Connect. The sync cursor advances to
max(currentPersisted, latestRecordTimestamp) instead of
sliceEndTimestamp. If no records were written for a slice, the cursor
stays put. This handles all late-arriving data generically without
needing per-data-type special-casing.
This commit is contained in:
Gideon Zenz
2026-03-02 20:19:02 +01:00
parent 1b314c41b1
commit b5b0b585cc
8 changed files with 166 additions and 13 deletions
@@ -396,7 +396,19 @@ class HealthConnectUtils {
recordsSyncedByType[key] = currentSynced + stat.recordsSynced
}
timestampToPersistForThisDataType = currentSliceEndTs
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) {
LOG.debug(
"$HC_SYNC_TAG Holding sync state for {}({}) at {} (no records synced in slice {} to {})",
gbDevice.aliasOrName, dataType.name, timestampToPersistForThisDataType,
currentSliceStartTs, currentSliceEndTs
)
}
currentSliceStartTs = currentSliceEndTs
} catch (e: SyncException) {
LOG.warn(
@@ -70,12 +70,11 @@ internal abstract class AbstractActivitySampleSyncer<TRecord : Record> : Activit
val recordsToInsert = mutableListOf<Record>()
var skippedCount = 0
var latestSyncedTimestamp: Instant? = null
relevantSamples.forEach { currentSample ->
val endTs = Instant.ofEpochSecond(currentSample.timestamp.toLong())
val startTs = endTs.minus(1, ChronoUnit.MINUTES)
// Use inclusive boundaries [sliceStart, sliceEnd] for the slice
// Overlap check: interval overlaps if endTs > sliceStart AND startTs <= sliceEnd
if (endTs.isBefore(sliceStartBoundary) || startTs.isAfter(sliceEndBoundary)) {
logger.debug(
"Skipping {} for device '{}' for sample at {} (interval {} to {}) as its interval is outside the slice {} - {}.",
@@ -96,6 +95,9 @@ internal abstract class AbstractActivitySampleSyncer<TRecord : Record> : Activit
return@forEach
}
recordsToInsert.add(record)
if (latestSyncedTimestamp == null || endTs.isAfter(latestSyncedTimestamp)) {
latestSyncedTimestamp = endTs
}
}
// 3. No Valid Records to Insert
@@ -114,7 +116,8 @@ internal abstract class AbstractActivitySampleSyncer<TRecord : Record> : Activit
return SyncerStatistics(
recordsSynced = recordsToInsert.size,
recordsSkipped = skippedCount,
recordType = recordTypeName
recordType = recordTypeName,
latestRecordTimestamp = latestSyncedTimestamp
)
}
}
@@ -99,10 +99,10 @@ internal abstract class AbstractTimeSampleSyncer<TSample : TimeSample, TRecord :
// Convert samples to records
var skippedCount = 0
var latestSyncedTimestamp: Instant? = null
val recordsToInsert = samples.filter {
val timestamp = Instant.ofEpochMilli(it.timestamp)
// Use inclusive boundaries [sliceStart, sliceEnd] to ensure last sample is included
if (timestamp.isBefore(sliceStartBoundary) || timestamp.isAfter(sliceEndBoundary)) {
logger.debug(
"Skipping sample for at {} as it's outside the slice {} - {}.",
@@ -113,13 +113,18 @@ internal abstract class AbstractTimeSampleSyncer<TSample : TimeSample, TRecord :
return@filter false
}
return@filter true
}.mapNotNull {
}.mapNotNull { sample ->
convertSample(
it,
sample,
offset,
metadata,
deviceName
) ?: run {
)?.also {
val ts = Instant.ofEpochMilli(sample.timestamp)
if (latestSyncedTimestamp == null || ts.isAfter(latestSyncedTimestamp)) {
latestSyncedTimestamp = ts
}
} ?: run {
skippedCount++
null
}
@@ -138,7 +143,8 @@ internal abstract class AbstractTimeSampleSyncer<TSample : TimeSample, TRecord :
return SyncerStatistics(
recordsSynced = recordsToInsert.size,
recordsSkipped = skippedCount,
recordType = recordTypeName
recordType = recordTypeName,
latestRecordTimestamp = latestSyncedTimestamp
)
}
}
@@ -30,7 +30,8 @@ import java.time.ZoneOffset
data class SyncerStatistics(
val recordsSynced: Int = 0,
val recordsSkipped: Int = 0,
val recordType: String = ""
val recordType: String = "",
val latestRecordTimestamp: Instant? = null
)
/**
@@ -143,6 +143,6 @@ internal object HeartRateSyncer : ActivitySampleSyncer {
}
LOG.info("Successfully inserted ${heartRateRecordList.size} HeartRateRecord(s) for device '$deviceName' for slice $sliceStartBoundary to $sliceEndBoundary.")
return SyncerStatistics(recordsSynced = heartRateRecordList.size, recordsSkipped = skippedCount, recordType = "HeartRate")
return SyncerStatistics(recordsSynced = heartRateRecordList.size, recordsSkipped = skippedCount, recordType = "HeartRate", latestRecordTimestamp = previousSampleTimestamp)
}
}
@@ -106,6 +106,7 @@ internal object RecordedWorkoutSyncer {
var workoutsProcessedInThisSlice = 0
val workoutRecordList = mutableListOf<Record>()
var latestWorkoutEndTs: Instant? = null
for (workout in workouts) {
try {
@@ -187,6 +188,10 @@ internal object RecordedWorkoutSyncer {
HealthConnectUtils.insertRecords(recordsToInsert, healthConnectClient)
LOG.info("Successfully inserted ${recordsToInsert.size} record(s) for workout (Type: ${activityKind}, Start: $workoutStartInstant) for device '$deviceName'.")
workoutRecordList.addAll(recordsToInsert)
val currentEnd = workoutEndInstant
if (latestWorkoutEndTs == null || currentEnd.isAfter(latestWorkoutEndTs)) {
latestWorkoutEndTs = currentEnd
}
} catch (e: Exception) {
LOG.error("Error processing workout for device '$deviceName'", e)
// Continue with next workout instead of failing entire sync
@@ -205,7 +210,7 @@ internal object RecordedWorkoutSyncer {
}
LOG.info("Successfully inserted ${workoutRecordList.size} ExerciseSessionRecord(s) for device '$deviceName' for slice $sliceStartBoundary to $sliceEndBoundary.")
return SyncerStatistics(recordsSynced = workoutRecordList.size, recordType = "Workout")
return SyncerStatistics(recordsSynced = workoutRecordList.size, recordType = "Workout", latestRecordTimestamp = latestWorkoutEndTs)
}
@@ -101,7 +101,8 @@ internal object SleepSyncer : ContextualActivitySampleSyncer {
LOG.info("Attempting to insert ${sleepSessionRecordList.size} SleepSessionRecord(s) for device '$deviceName' for slice $sliceStartBoundary to $sliceEndBoundary.")
HealthConnectUtils.insertRecords(sleepSessionRecordList, healthConnectClient)
LOG.info("Successfully inserted SleepSessionRecord(s) for device '$deviceName' for slice $sliceStartBoundary to $sliceEndBoundary.")
return SyncerStatistics(recordsSynced = sleepSessionRecordList.size, recordsSkipped = skippedCount, recordType = "Sleep")
val latestTs = sleepSessionRecordList.maxOfOrNull { it.endTime }
return SyncerStatistics(recordsSynced = sleepSessionRecordList.size, recordsSkipped = skippedCount, recordType = "Sleep", latestRecordTimestamp = latestTs)
}
/**
@@ -0,0 +1,125 @@
package nodomain.freeyourgadget.gadgetbridge.util.healthconnect
import nodomain.freeyourgadget.gadgetbridge.util.healthconnect.syncers.SyncerStatistics
import org.junit.Assert.*
import org.junit.Test
import java.time.Instant
class SyncTimestampAdvancementTest {
private val baseEpoch = 1_700_000_000L
private val baseInstant: Instant = Instant.ofEpochSecond(baseEpoch)
@Test
fun latestRecordTimestamp_advancesCursor_whenPresent() {
val stats = listOf(
SyncerStatistics(recordsSynced = 3, recordType = "Steps", latestRecordTimestamp = baseInstant.plusSeconds(3600))
)
val latestRecordTs = stats.mapNotNull { it.latestRecordTimestamp }.maxOrNull()
assertNotNull(latestRecordTs)
assertTrue(latestRecordTs!!.isAfter(baseInstant))
}
@Test
fun cursorDoesNotAdvance_whenNoRecordsSynced() {
val stats = listOf(
SyncerStatistics(recordsSynced = 0, recordType = "Sleep")
)
val latestRecordTs = stats.mapNotNull { it.latestRecordTimestamp }.maxOrNull()
assertNull(latestRecordTs)
}
@Test
fun multipleStats_usesMaxTimestamp() {
val earlier = baseInstant.plusSeconds(1000)
val later = baseInstant.plusSeconds(5000)
val stats = listOf(
SyncerStatistics(recordsSynced = 2, recordType = "Steps", latestRecordTimestamp = earlier),
SyncerStatistics(recordsSynced = 5, recordType = "HeartRate", latestRecordTimestamp = later),
SyncerStatistics(recordsSynced = 1, recordType = "Calories", latestRecordTimestamp = earlier)
)
val latestRecordTs = stats.mapNotNull { it.latestRecordTimestamp }.maxOrNull()
assertEquals(later, latestRecordTs)
}
@Test
fun mixedNullAndNonNull_usesNonNull() {
val ts = baseInstant.plusSeconds(7200)
val stats = listOf(
SyncerStatistics(recordsSynced = 0, recordType = "Steps"),
SyncerStatistics(recordsSynced = 1, recordType = "HeartRate", latestRecordTimestamp = ts),
SyncerStatistics(recordsSynced = 0, recordType = "Calories")
)
val latestRecordTs = stats.mapNotNull { it.latestRecordTimestamp }.maxOrNull()
assertEquals(ts, latestRecordTs)
}
@Test
fun allNullTimestamps_cursorStaysAtCurrentValue() {
val currentPersisted = baseInstant
val stats = listOf(
SyncerStatistics(recordsSynced = 0, recordType = "Sleep"),
SyncerStatistics(recordsSynced = 0, recordType = "Workout")
)
val latestRecordTs = stats.mapNotNull { it.latestRecordTimestamp }.maxOrNull()
val newPersisted = if (latestRecordTs != null && latestRecordTs.isAfter(currentPersisted)) {
latestRecordTs
} else {
currentPersisted
}
assertEquals(currentPersisted, newPersisted)
}
@Test
fun cursorDoesNotGoBackward_whenLatestIsBeforeCurrent() {
val currentPersisted = baseInstant.plusSeconds(10000)
val olderTs = baseInstant.plusSeconds(5000)
val stats = listOf(
SyncerStatistics(recordsSynced = 1, recordType = "Steps", latestRecordTimestamp = olderTs)
)
val latestRecordTs = stats.mapNotNull { it.latestRecordTimestamp }.maxOrNull()
val newPersisted = if (latestRecordTs != null && latestRecordTs.isAfter(currentPersisted)) {
latestRecordTs
} else {
currentPersisted
}
assertEquals(currentPersisted, newPersisted)
}
@Test
fun emptyStatsList_cursorStays() {
val currentPersisted = baseInstant
val stats = emptyList<SyncerStatistics>()
val latestRecordTs = stats.mapNotNull { it.latestRecordTimestamp }.maxOrNull()
val newPersisted = if (latestRecordTs != null && latestRecordTs.isAfter(currentPersisted)) {
latestRecordTs
} else {
currentPersisted
}
assertEquals(currentPersisted, newPersisted)
}
@Test
fun defaultSyncerStatistics_hasNullTimestamp() {
val stats = SyncerStatistics()
assertNull(stats.latestRecordTimestamp)
}
@Test
fun syncerStatistics_preservesTimestampInCopy() {
val ts = Instant.ofEpochSecond(baseEpoch + 999)
val original = SyncerStatistics(recordsSynced = 1, recordType = "HRV", latestRecordTimestamp = ts)
val copy = original.copy(recordsSynced = 2)
assertEquals(ts, copy.latestRecordTimestamp)
assertEquals(2, copy.recordsSynced)
}
}