Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .github/nightly-notes/2026-10-29-fix-sync-clock.md
Original file line number Diff line number Diff line change
@@ -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
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -36,6 +37,7 @@ class SyncChangeTracker
constructor(
private val data: SyncDataAccess,
private val store: SyncChangeLedgerStore,
private val identity: DeviceIdentity,
) {
private val json =
Json {
Expand All @@ -49,7 +51,8 @@ class SyncChangeTracker

suspend fun watchHistory(node: String): List<CanonicalWatchHistory> {
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())
Expand All @@ -68,7 +71,8 @@ class SyncChangeTracker

suspend fun likes(node: String): List<CanonicalLike> {
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)
Expand All @@ -88,7 +92,8 @@ class SyncChangeTracker

suspend fun notes(): List<CanonicalNote> {
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)
Expand All @@ -108,7 +113,8 @@ class SyncChangeTracker

suspend fun groups(node: String): List<CanonicalSubscriptionGroup> {
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())
Expand All @@ -128,11 +134,12 @@ class SyncChangeTracker

suspend fun playlists(node: String): List<CanonicalPlaylist> {
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 ->
Expand Down Expand Up @@ -221,7 +228,8 @@ class SyncChangeTracker

suspend fun settings(node: String): List<CanonicalSetting> {
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
Expand Down Expand Up @@ -263,6 +271,7 @@ class SyncChangeTracker
templates: Map<String, String> = emptyMap(),
) {
store.update(collection) { it.settle(current, stamps, peerTombstones, templates) }
(stamps.values + peerTombstones.values.map { it.atMs }).maxOrNull()?.let { identity.observe(it) }
}

private fun <T> peer(
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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())

Expand All @@ -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")
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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<Long>()
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()) }
Expand Down Expand Up @@ -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)
}
}
Loading