diff --git a/.github/nightly-notes/2026-10-29-fix-sync-clock.md b/.github/nightly-notes/2026-10-29-fix-sync-clock.md new file mode 100644 index 0000000000..d78eaf8f0f --- /dev/null +++ b/.github/nightly-notes/2026-10-29-fix-sync-clock.md @@ -0,0 +1,2 @@ +## Fixed +- A change you make right after syncing is no longer undone by the other device when this one's clock runs behind diff --git a/app/src/main/java/io/github/aedev/flow/sync/apply/SyncChangeTracker.kt b/app/src/main/java/io/github/aedev/flow/sync/apply/SyncChangeTracker.kt index 7a32b803d1..e287c7d43a 100644 --- a/app/src/main/java/io/github/aedev/flow/sync/apply/SyncChangeTracker.kt +++ b/app/src/main/java/io/github/aedev/flow/sync/apply/SyncChangeTracker.kt @@ -15,6 +15,7 @@ import io.github.aedev.flow.sync.canonical.CanonicalPlaylistItem import io.github.aedev.flow.sync.canonical.CanonicalSetting import io.github.aedev.flow.sync.canonical.CanonicalSubscriptionGroup import io.github.aedev.flow.sync.canonical.CanonicalWatchHistory +import io.github.aedev.flow.sync.identity.DeviceIdentity import io.github.aedev.flow.sync.identity.Hlc import io.github.aedev.flow.sync.mapping.PlaylistMapper import io.github.aedev.flow.sync.mapping.SettingsMapper @@ -36,6 +37,7 @@ class SyncChangeTracker constructor( private val data: SyncDataAccess, private val store: SyncChangeLedgerStore, + private val identity: DeviceIdentity, ) { private val json = Json { @@ -49,7 +51,8 @@ class SyncChangeTracker suspend fun watchHistory(node: String): List { val live = data.readWatchHistory(node) - val ledger = store.update(WATCH_HISTORY) { it.observe(live.associate { r -> r.videoId to PRESENT }, now()) } + val at = now() + val ledger = store.update(WATCH_HISTORY) { it.observe(live.associate { r -> r.videoId to PRESENT }, at) } return live + ledger.tombstones().map { (videoId, e) -> peerRecord(e, CanonicalWatchHistory.serializer()) @@ -68,7 +71,8 @@ class SyncChangeTracker suspend fun likes(node: String): List { val live = data.readLikes(node) - val ledger = store.update(LIKES) { it.observe(live.associate { r -> likeKey(r) to PRESENT }, now()) } + val at = now() + val ledger = store.update(LIKES) { it.observe(live.associate { r -> likeKey(r) to PRESENT }, at) } return live + ledger.tombstones().mapNotNull { (key, e) -> peerRecord(e, CanonicalLike.serializer()) ?: likeTombstone(key, e.atMs, node) @@ -88,7 +92,8 @@ class SyncChangeTracker suspend fun notes(): List { val live = data.readNotes() - val ledger = store.update(NOTES) { it.observe(live.associate { n -> n.id to PRESENT }, now()) } + val at = now() + val ledger = store.update(NOTES) { it.observe(live.associate { n -> n.id to PRESENT }, at) } return live + ledger.tombstones().map { (id, e) -> peerRecord(e, CanonicalNote.serializer()) ?: CanonicalNote(id = id, updatedAt = e.atMs, deleted = true) @@ -108,7 +113,8 @@ class SyncChangeTracker suspend fun groups(node: String): List { val live = data.readSubscriptions(hlc = "") - val ledger = store.update(GROUPS) { it.observe(live.associate { g -> g.name to groupHash(g) }, now()) } + val at = now() + val ledger = store.update(GROUPS) { it.observe(live.associate { g -> g.name to groupHash(g) }, at) } return live.map { it.copy(hlc = stamp(ledger.stampOf(it.name), node)) } + ledger.tombstones().map { (name, e) -> peerRecord(e, CanonicalSubscriptionGroup.serializer()) @@ -128,11 +134,12 @@ class SyncChangeTracker suspend fun playlists(node: String): List { val live = data.readPlaylists(hlc = "") + val at = now() val playlists = store.update(PLAYLISTS) { - it.observe(live.associate { p -> p.syncId to playlistHash(p) }, now(), templates(live)) + it.observe(live.associate { p -> p.syncId to playlistHash(p) }, at, templates(live)) } - val items = store.update(PLAYLIST_ITEMS) { it.observe(itemHashes(live), now()) } + val items = store.update(PLAYLIST_ITEMS) { it.observe(itemHashes(live), at) } val liveIds = live.mapTo(HashSet()) { it.syncId } val stamped = live.map { p -> @@ -221,7 +228,8 @@ class SyncChangeTracker suspend fun settings(node: String): List { val live = data.readSettings(hlc = "") - val ledger = store.update(SETTINGS) { it.observe(live.associate { s -> s.key to s.value.toString().hashCode() }, now()) } + val at = now() + val ledger = store.update(SETTINGS) { it.observe(live.associate { s -> s.key to s.value.toString().hashCode() }, at) } val liveKeys = live.mapTo(HashSet()) { it.key } val relayed = store @@ -263,6 +271,7 @@ class SyncChangeTracker templates: Map = emptyMap(), ) { store.update(collection) { it.settle(current, stamps, peerTombstones, templates) } + (stamps.values + peerTombstones.values.map { it.atMs }).maxOrNull()?.let { identity.observe(it) } } private fun peer( @@ -290,7 +299,7 @@ class SyncChangeTracker return CanonicalLike(kind = kind, id = id, state = CanonicalLike.STATE_NONE, updatedAtMs = atMs, hlc = stamp(atMs, node)) } - private fun now() = System.currentTimeMillis() + private suspend fun now() = identity.editTime() private companion object { const val PRESENT = 1 diff --git a/app/src/main/java/io/github/aedev/flow/sync/identity/DeviceIdentity.kt b/app/src/main/java/io/github/aedev/flow/sync/identity/DeviceIdentity.kt index 0d0957d141..008c23c752 100644 --- a/app/src/main/java/io/github/aedev/flow/sync/identity/DeviceIdentity.kt +++ b/app/src/main/java/io/github/aedev/flow/sync/identity/DeviceIdentity.kt @@ -3,6 +3,7 @@ package io.github.aedev.flow.sync.identity import android.content.Context import android.os.Build import androidx.datastore.preferences.core.edit +import androidx.datastore.preferences.core.longPreferencesKey import androidx.datastore.preferences.core.stringPreferencesKey import dagger.hilt.android.qualifiers.ApplicationContext import io.github.aedev.flow.BuildConfig @@ -45,6 +46,17 @@ class DeviceIdentity store.edit { it[KEY_DEVICE_NAME] = name.trim().take(64) } } + /** + * When a local edit happens, for its stamp: never before a stamp this device has applied, so an + * edit made after reading another device's change wins even when this clock is behind. + */ + suspend fun editTime(): Long = maxOf(System.currentTimeMillis(), (store.data.first()[KEY_CLOCK] ?: 0L) + 1) + + /** Remembers [physicalMs] if it is the newest time among the stamps this device has applied. */ + suspend fun observe(physicalMs: Long) { + store.edit { if (physicalMs > (it[KEY_CLOCK] ?: 0L)) it[KEY_CLOCK] = physicalMs } + } + /** The 8-char HLC node id for this device. */ suspend fun hlcNode(): String = Hlc.nodeFromDeviceId(deviceId()) @@ -59,5 +71,6 @@ class DeviceIdentity companion object { private val KEY_DEVICE_ID = stringPreferencesKey("device_id") private val KEY_DEVICE_NAME = stringPreferencesKey("device_name") + private val KEY_CLOCK = longPreferencesKey("clock_ms") } } diff --git a/app/src/test/java/io/github/aedev/flow/sync/apply/SyncChangeTrackerTest.kt b/app/src/test/java/io/github/aedev/flow/sync/apply/SyncChangeTrackerTest.kt index 911fbae333..c9f9d41b2b 100644 --- a/app/src/test/java/io/github/aedev/flow/sync/apply/SyncChangeTrackerTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync/apply/SyncChangeTrackerTest.kt @@ -6,6 +6,7 @@ import io.github.aedev.flow.sync.canonical.CanonicalPlaylistItem import io.github.aedev.flow.sync.canonical.CanonicalSetting import io.github.aedev.flow.sync.canonical.CanonicalSubscriptionGroup import io.github.aedev.flow.sync.canonical.CanonicalWatchHistory +import io.github.aedev.flow.sync.identity.DeviceIdentity import io.github.aedev.flow.sync.identity.Hlc import io.mockk.coEvery import io.mockk.mockk @@ -29,7 +30,14 @@ class SyncChangeTrackerTest { } coEvery { load(capture(name)) } answers { ledgers[name.captured] ?: CollectionLedger() } } - private val tracker = SyncChangeTracker(data, store) + private var clock = 0L + private val identity: DeviceIdentity = + mockk { + coEvery { editTime() } answers { maxOf(System.currentTimeMillis(), clock + 1) } + val seen = slot() + coEvery { observe(capture(seen)) } answers { clock = maxOf(clock, seen.captured) } + } + private val tracker = SyncChangeTracker(data, store, identity) private fun history(vararg ids: String) = ids.map { CanonicalWatchHistory(videoId = it, watchedAtMs = 10, hlc = Hlc(10, 0, node).encode()) } @@ -139,4 +147,18 @@ class SyncChangeTrackerTest { assertEquals(desktopOnly, out.single { it.key == desktopOnly.key }) assertEquals(0L, Hlc.decode(out.single { it.key == "autoplay" }.hlc).physicalMs) } + + @Test + fun an_edit_after_a_newer_peer_stamp_is_stamped_after_it() = + runBlocking { + val later = System.currentTimeMillis() + 3_600_000 + val group = CanonicalSubscriptionGroup(name = "Music", channelIds = listOf("c1"), hlc = Hlc(later, 0, "deskuuid").encode()) + coEvery { data.readSubscriptions("") } returns listOf(group.copy(hlc = "")) + tracker.settleGroups(listOf(group)) + coEvery { data.readSubscriptions("") } returns listOf(group.copy(channelIds = listOf("c1", "c2"), hlc = "")) + + val edited = tracker.groups(node).single() + + assertTrue("a clock an hour behind still stamps after the peer", Hlc.decode(edited.hlc).physicalMs > later) + } }