diff --git a/app/src/main/java/com/pombo/android/ChannelManager.kt b/app/src/main/java/com/pombo/android/ChannelManager.kt index e63d0ca..53e562e 100644 --- a/app/src/main/java/com/pombo/android/ChannelManager.kt +++ b/app/src/main/java/com/pombo/android/ChannelManager.kt @@ -777,10 +777,23 @@ class ChannelManager( // the store on the next save. internal val _channels = MutableStateFlow(store.load().distinctBy { it.messageStreamId }) + /** + * Each record as last persisted or imported: what [saveChannels] compares + * against to stamp the fields that changed here. + */ + @Volatile private var persisted: Map = _channels.value.associateBy { it.messageStreamId } + + /** Writes the list without stamping it: loads, imports and derived repairs. */ + private fun persist(channels: List) { + store.save(channels) + persisted = channels.associateBy { it.messageStreamId } + } + /** Re-reads the list after the storage scope changes (account switch/guest). */ fun reloadChannels() { scope.launch { channelSwitchMutex.withLock { closeCurrentInternal(); clearReopenSnapshots() } } _channels.value = store.load().distinctBy { it.messageStreamId } + persisted = _channels.value.associateBy { it.messageStreamId } _channelOrder.value = store.loadOrder() } @@ -1287,12 +1300,11 @@ class ChannelManager( com.pombo.android.core.GraphApi.getChannelInfo(channel.messageStreamId) } catch (e: Exception) { null } ?: return false if (info.name.isNullOrEmpty()) return false - val fixed = channel.copy(exposure = info.exposure.ifEmpty { "visible" }) - _channels.value = _channels.value.map { - if (it.messageStreamId == fixed.messageStreamId) fixed else it + val exposure = info.exposure.ifEmpty { "visible" } + if (updateStored(channel.messageStreamId) { it.copy(exposure = exposure) } == null) { + _current.value?.takeIf { it.messageStreamId == channel.messageStreamId } + ?.let { _current.value = it.copy(exposure = exposure) } } - saveChannels() - if (_current.value?.messageStreamId == fixed.messageStreamId) _current.value = fixed Log.i(TAG, "rename: chain says this channel is named publicly — exposure corrected") return true } @@ -1312,16 +1324,16 @@ class ChannelManager( bridge.call("updateStreamMetadata", args, 120_000) } - val updated = channel.copy( - name = name?.trim()?.ifEmpty { null } ?: channel.name, - description = description?.trim() ?: channel.description, - // Stamped so the Graph refresh below does not revert this edit while - // the subgraph is still catching up. - metaUpdatedAt = System.currentTimeMillis() - ) - _channels.value = _channels.value.map { if (it.messageStreamId == updated.messageStreamId) updated else it } - saveChannels() - _current.value = updated + val editedAt = System.currentTimeMillis() + updateStored(channel.messageStreamId) { + it.copy( + name = name?.trim()?.ifEmpty { null } ?: it.name, + description = description?.trim() ?: it.description, + // Stamped so the Graph refresh below does not revert this edit while + // the subgraph is still catching up. + metaUpdatedAt = editedAt + ) + } } /** @@ -1332,15 +1344,20 @@ class ChannelManager( * falls back to. */ suspend fun refreshChannelMetadataFromGraph(): Boolean { - var changed = false - val updates = mutableMapOf() + val infos = mutableMapOf() for (channel in _channels.value) { if (channel.type == "dm") continue - val info = try { + infos[channel.messageStreamId] = try { com.pombo.android.core.GraphApi.getChannelInfo(channel.messageStreamId) } catch (e: Exception) { null } ?: continue + } + // Compared with the records as they are after the lookups, which may + // have run long enough for a sync import or a local edit to land. + val updates = mutableMapOf() + for (channel in _channels.value) { + val info = infos[channel.messageStreamId] ?: continue // Indexing lag: never overwrite a local admin edit with older data. if (channel.metaUpdatedAt != null && info.updatedAt <= channel.metaUpdatedAt) continue @@ -1359,13 +1376,18 @@ class ChannelManager( if (info.exposure == "visible" && info.description != channel.description) { next = next.copy(description = info.description) } - if (next !== channel) { updates[channel.messageStreamId] = next; changed = true } + if (next !== channel) updates[channel.messageStreamId] = next } + val changed = updates.isNotEmpty() if (changed) { _channels.value = _channels.value.map { updates[it.messageStreamId] ?: it } saveChannels() // Keep the open channel's header in sync with the list. - _current.value?.let { cur -> updates[cur.messageStreamId]?.let { _current.value = it } } + _current.value?.let { cur -> + if (cur.messageStreamId in updates) { + _channels.value.find { it.messageStreamId == cur.messageStreamId }?.let { _current.value = it } + } + } } return changed } @@ -1775,20 +1797,15 @@ class ChannelManager( // with the old retention. A stream already at `days` counts as in // sync too, which is the whole point of not writing to it. fun settled(kind: String) = out.results[kind] == "applied" || out.results[kind] == "unchanged" - val updated = channel.copy( - storageDays = if (settled("message")) days else channel.storageDays, - adminStorageDays = if (settled("admin")) days else channel.adminStorageDays, - keysStorageDays = if (settled("keys")) days else channel.keysStorageDays, + val latest = _channels.value.find { it.messageStreamId == channel.messageStreamId } ?: return out + val updated = latest.copy( + storageDays = if (settled("message")) days else latest.storageDays, + adminStorageDays = if (settled("admin")) days else latest.adminStorageDays, + keysStorageDays = if (settled("keys")) days else latest.keysStorageDays, interactionsStorageDays = - if (settled("interactions")) days else channel.interactionsStorageDays + if (settled("interactions")) days else latest.interactionsStorageDays ) - if (updated != channel) { - _channels.value = _channels.value.map { - if (it.messageStreamId == updated.messageStreamId) updated else it - } - saveChannels() - if (_current.value?.messageStreamId == updated.messageStreamId) _current.value = updated - } + if (updated != latest) updateStored(channel.messageStreamId) { updated } return out } @@ -4687,7 +4704,7 @@ class ChannelManager( } // Local only: these ids derive from the stream id, and the web never stores // keysStreamId, so a sync push here would follow each of its pulls. - store.save(_channels.value) + persist(_channels.value) } return healed } @@ -5301,7 +5318,7 @@ class ChannelManager( /** Replaces the channel list wholesale after a sync merge. */ fun replaceChannels(channels: List) { _channels.value = channels - store.save(channels) + persist(channels) reconcileAllGateAuthority() } @@ -7183,9 +7200,31 @@ class ChannelManager( saveChannels() } - /** Every local change goes through here; a sync import must not, or each pull is pushed back out. */ + /** + * Changes the stored record as it is now, not a copy captured before a + * slow call: a sync import may have landed meanwhile, and the save stamps + * every field that differs, so a stale copy would win them all on every + * device. Null when the channel is not stored. + */ + internal fun updateStored(messageStreamId: String, change: (Channel) -> Channel): Channel? { + val latest = _channels.value.find { it.messageStreamId == messageStreamId } ?: return null + val updated = change(latest) + _channels.value = _channels.value.map { if (it.messageStreamId == messageStreamId) updated else it } + saveChannels() + val stored = _channels.value.find { it.messageStreamId == messageStreamId } ?: updated + if (_current.value?.messageStreamId == messageStreamId) _current.value = stored + return stored + } + + /** + * Every local change goes through here; a sync import must not, or each + * pull is pushed back out, and its fields would be stamped as changed here. + */ internal fun saveChannels() { - store.save(_channels.value) + val now = System.currentTimeMillis() + val stamped = _channels.value.map { it.stampedAgainst(persisted[it.messageStreamId], now) } + _channels.value = stamped + persist(stamped) onLocalStateChanged() } diff --git a/app/src/main/java/com/pombo/android/core/SyncMerge.kt b/app/src/main/java/com/pombo/android/core/SyncMerge.kt index 805abae..82d7290 100644 --- a/app/src/main/java/com/pombo/android/core/SyncMerge.kt +++ b/app/src/main/java/com/pombo/android/core/SyncMerge.kt @@ -135,7 +135,8 @@ object SyncMerge { * LWW-element-set over channels. Membership is decided by comparing the * join timestamp against the leave tombstone; the most recent action wins * and a tie goes to Join. Snapshot replacement would let an older remote - * snapshot delete a channel this device just joined. + * snapshot delete a channel this device just joined. Two copies of the + * same channel merge field by field ([mergeChannelRecord]). */ private fun mergeChannels( baseChannels: JSONArray?, @@ -170,7 +171,11 @@ object SyncMerge { val id = c.optStringOrNull("messageStreamId") ?: continue noteJoin(id, c) val existing = byId[id] - if (existing == null || replaces(c, existing)) byId[id] = c + byId[id] = when { + existing == null -> c + replaces(c, existing) -> mergeChannelRecord(c, existing) + else -> mergeChannelRecord(existing, c) + } } } @@ -185,14 +190,53 @@ object SyncMerge { return channels to leftAt } + private fun fieldStamp(record: JSONObject, key: String): Long = + (record.optJSONObject("fieldTs")?.opt(key) as? Number)?.toLong() ?: 0L + + /** + * One channel record from two copies of it (web mergeChannelRecord). Each + * field comes from the copy whose `fieldTs` stamped it later; a field + * neither copy stamped, or both stamped at the same time, comes from + * [preferred], and a field only one copy carries is kept. The stamps join, + * the latest per field, so a client that drops them cannot take them from + * the others. Returns [preferred] itself when nothing comes from [other]. + */ + fun mergeChannelRecord(preferred: JSONObject, other: JSONObject): JSONObject { + val merged = JSONObject() + var tookOther = false + val keys = LinkedHashSet() + preferred.keys().forEach { keys.add(it) } + other.keys().forEach { keys.add(it) } + for (key in keys) { + if (key == "fieldTs") continue + val fromOther = other.has(key) && + (!preferred.has(key) || fieldStamp(other, key) > fieldStamp(preferred, key)) + merged.put(key, if (fromOther) other.get(key) else preferred.get(key)) + if (fromOther) tookOther = true + } + if (!tookOther) return preferred + val sources = listOfNotNull(preferred.optJSONObject("fieldTs"), other.optJSONObject("fieldTs")) + if (sources.isNotEmpty()) { + val stamps = JSONObject() + sources.forEach { src -> + src.keys().forEach { key -> + val ts = (src.opt(key) as? Number)?.toLong() ?: return@forEach + if (ts > stamps.optLong(key, 0L)) stamps.put(key, ts) + } + } + merged.put("fieldTs", stamps) + } + return merged + } + private fun joinTs(channel: JSONObject): Long { val joined = channel.optLong("joinedAt", 0L) return if (joined > 0L) joined else channel.optLong("createdAt", 0L) } - /** Incoming wins ties: the newer snapshot carries fresher metadata. An - * entry with no joinedAt of its own is never a newer join than one that - * has it, but its time still counts as a join against a leave tombstone. */ + /** For the fields no stamp decides: incoming wins ties. An entry with no + * joinedAt of its own is never a newer join than one that has it, but its + * time still counts as a join against a leave tombstone. */ private fun replaces(incoming: JSONObject, existing: JSONObject): Boolean { val incomingJoined = incoming.optLong("joinedAt", 0L) > 0L val existingJoined = existing.optLong("joinedAt", 0L) > 0L diff --git a/app/src/main/java/com/pombo/android/core/channels/Moderation.kt b/app/src/main/java/com/pombo/android/core/channels/Moderation.kt index de0426c..3fc0f19 100644 --- a/app/src/main/java/com/pombo/android/core/channels/Moderation.kt +++ b/app/src/main/java/com/pombo/android/core/channels/Moderation.kt @@ -99,10 +99,11 @@ internal class Moderation(private val manager: ChannelManager) { val keysId = channel.keysStreamId.ifEmpty { StreamConstants.deriveKeysId(messageStreamId) } epochKeys.rotateEpoch(messageStreamId, keysId) } - override suspend fun covered(messageStreamId: String, addresses: Set) = + override suspend fun covered(messageStreamId: String, addresses: Set) { updateStored(messageStreamId) { it.copy(rotatedForNoAccess = (it.rotatedForNoAccess + addresses).distinct()) } + } override fun stillOwned(messageStreamId: String): Boolean = _channels.value.find { it.messageStreamId == messageStreamId }?.let { amOwner(it) } == true override fun loadOwed(key: String): List { @@ -122,13 +123,8 @@ internal class Moderation(private val manager: ChannelManager) { _channels.value.filter { it.type == "gated" && amOwner(it) }.map { it.messageStreamId }) /** Change the stored record as it is now, not a copy captured before a slow call. */ - private fun updateStored(messageStreamId: String, change: (Channel) -> Channel) { - val latest = _channels.value.find { it.messageStreamId == messageStreamId } ?: return - val updated = change(latest) - _channels.value = _channels.value.map { if (it.messageStreamId == messageStreamId) updated else it } - manager.saveChannels() - if (_current.value?.messageStreamId == messageStreamId) _current.value = updated - } + private fun updateStored(messageStreamId: String, change: (Channel) -> Channel): Channel? = + manager.updateStored(messageStreamId, change) internal val _pins = MutableStateFlow>(emptyList()) @@ -546,10 +542,8 @@ internal class Moderation(private val manager: ChannelManager) { val gate = channel.gateAddress ?: throw IllegalStateException("Gate address unknown (repair pending)") bridge.call("gateAllow", JSONObject().put("gate", gate).put("user", addr), 180_000) - val updated = channel.copy(members = channel.members + addr) - _channels.value = _channels.value.map { if (it.messageStreamId == updated.messageStreamId) updated else it } - manager.saveChannels() - _current.value = updated + val updated = updateStored(channel.messageStreamId) { withMember(it, addr) } + ?: withMember(channel, addr) answerWaitingRequests(updated) return } @@ -572,12 +566,14 @@ internal class Moderation(private val manager: ChannelManager) { setPermissionsRetry(keysId, rw) com.pombo.android.core.GraphApi.clearCache() - val updated = channel.copy(members = channel.members + addr) - _channels.value = _channels.value.map { if (it.messageStreamId == updated.messageStreamId) updated else it } - manager.saveChannels() - _current.value = updated + updateStored(channel.messageStreamId) { withMember(it, addr) } } + /** [channel] with [addr] among its members, once. */ + private fun withMember(channel: Channel, addr: String): Channel = + if (channel.members.any { it.equals(addr, ignoreCase = true) }) channel + else channel.copy(members = channel.members + addr) + /** * Revokes all permissions (web: empty permission array = revoke). * @return false when the key rotation that follows is still owed @@ -624,10 +620,9 @@ internal class Moderation(private val manager: ChannelManager) { setPermissionsRetry(keysId, revoke) com.pombo.android.core.GraphApi.clearCache() - val updated = channel.copy(members = channel.members.filterNot { it.equals(addr, ignoreCase = true) }) - _channels.value = _channels.value.map { if (it.messageStreamId == updated.messageStreamId) updated else it } - manager.saveChannels() - _current.value = updated + updateStored(channel.messageStreamId) { latest -> + latest.copy(members = latest.members.filterNot { it.equals(addr, ignoreCase = true) }) + } // Rotate the epoch so the removed member cannot read anything published // from here on — they keep what they already read; the rotation protects diff --git a/app/src/main/java/com/pombo/android/data/Channel.kt b/app/src/main/java/com/pombo/android/data/Channel.kt index 206fffd..86d76f8 100644 --- a/app/src/main/java/com/pombo/android/data/Channel.kt +++ b/app/src/main/java/com/pombo/android/data/Channel.kt @@ -87,8 +87,31 @@ data class Channel( * a lag, so a refresh whose `updatedAt` predates this would revert a rename * the admin just made (web: `channel.metaUpdatedAt`). */ - val metaUpdatedAt: Long? = null + val metaUpdatedAt: Long? = null, + /** + * When each field (by its JSON key) last changed on some device. The sync + * merges the record field by field on it (web: `channel.fieldTs`). + */ + val fieldTs: Map = emptyMap() ) { + /** + * This record with every field that differs from [previous], the copy + * last persisted or imported, stamped [now] (web stampChangedFields). A + * record with no such copy (created here) keeps the stamps it has; a stamp + * never goes back, whichever copy holds it. + */ + fun stampedAgainst(previous: Channel?, now: Long): Channel { + if (previous == null || previous == this) return this + val stamps = HashMap(previous.fieldTs) + fieldTs.forEach { (key, ts) -> if (ts > (stamps[key] ?: 0L)) stamps[key] = ts } + val mine = toJson() + val before = previous.toJson() + (mine.keys().asSequence() + before.keys().asSequence()).toSet().forEach { key -> + if (key != "fieldTs" && mine.opt(key)?.toString() != before.opt(key)?.toString()) stamps[key] = now + } + return if (stamps == fieldTs) this else copy(fieldTs = stamps) + } + fun toJson(): JSONObject = JSONObject() .put("peerAddress", peerAddress ?: JSONObject.NULL) .put("messageStreamId", messageStreamId) @@ -124,6 +147,7 @@ data class Channel( .put("readOnly", readOnly) .put("writeOnly", writeOnly) .put("metaUpdatedAt", metaUpdatedAt ?: JSONObject.NULL) + .apply { if (fieldTs.isNotEmpty()) put("fieldTs", JSONObject(fieldTs)) } companion object { fun fromJson(o: JSONObject): Channel { @@ -193,7 +217,12 @@ data class Channel( readOnly = o.optBoolean("readOnly", false), writeOnly = o.optBoolean("writeOnly", false), peerAddress = if (o.isNull("peerAddress")) null else o.optString("peerAddress").ifEmpty { null }, - metaUpdatedAt = if (o.isNull("metaUpdatedAt")) null else o.optLong("metaUpdatedAt") + metaUpdatedAt = if (o.isNull("metaUpdatedAt")) null else o.optLong("metaUpdatedAt"), + fieldTs = o.optJSONObject("fieldTs")?.let { s -> + s.keys().asSequence() + .mapNotNull { key -> (s.opt(key) as? Number)?.let { key to it.toLong() } } + .toMap() + } ?: emptyMap() ) } } diff --git a/app/src/test/java/com/pombo/android/AdminPollProbeTest.kt b/app/src/test/java/com/pombo/android/AdminPollProbeTest.kt index f7fdfe5..6c9bc64 100644 --- a/app/src/test/java/com/pombo/android/AdminPollProbeTest.kt +++ b/app/src/test/java/com/pombo/android/AdminPollProbeTest.kt @@ -60,8 +60,11 @@ class AdminPollProbeTest { @After fun tearDown() = h.stop() + /** Snapshots are recent, or the owner's open republishes them as nearing their TTL. */ + private val t0 = System.currentTimeMillis() + private fun snapshot(rev: Int, hidden: Int = 5) = JSONObject() - .put("type", "ADMIN_STATE").put("rev", rev).put("ts", 1_000L * rev).put("createdBy", me) + .put("type", "ADMIN_STATE").put("rev", rev).put("ts", t0 + rev).put("createdBy", me) .put("state", JSONObject() .put("bannedMembers", JSONArray()) .put("hiddenMessageIds", JSONArray((0 until hidden).map { "message-id-%030d".format(it) })) diff --git a/app/src/test/java/com/pombo/android/AdminStateSplitTest.kt b/app/src/test/java/com/pombo/android/AdminStateSplitTest.kt index 1d3eb44..bd6d1a3 100644 --- a/app/src/test/java/com/pombo/android/AdminStateSplitTest.kt +++ b/app/src/test/java/com/pombo/android/AdminStateSplitTest.kt @@ -27,8 +27,13 @@ class AdminStateSplitTest { .address("0x4f3edf983ac636a65a842ce7c78d9aa706d3b113bce9c46f30d7d21715b23b1d").lowercase() private val streamId = "$me/room-1" private val room = ChannelManagerHarness.channel(streamId) - private val h = ChannelManagerHarness(channels = listOf(room)) + /** A room this account only reads: where a member meets the signal. */ + private val owner = "0x" + "ab".repeat(20) + private val theirs = ChannelManagerHarness.channel("$owner/room-1") + private val h = ChannelManagerHarness(channels = listOf(room, theirs)) private val manager = h.manager + /** Snapshots are recent, or the owner's open republishes them as nearing their TTL. */ + private val t0 = System.currentTimeMillis() /** The -3 as the node holds it, oldest first: (row, publisher). */ private val storage = mutableListOf>() @@ -44,7 +49,7 @@ class AdminStateSplitTest { coEvery { h.bridge.call("resend", any(), any()) } answers { val args = secondArg() val messages = JSONArray() - if (args.optString("streamId") == room.adminStreamId) { + if (args.optString("streamId") in setOf(room.adminStreamId, theirs.adminStreamId)) { val last = args.optInt("last") reads += last for ((row, publisher) in storage.takeLast(last)) { @@ -58,8 +63,8 @@ class AdminStateSplitTest { @After fun tearDown() = h.stop() - private fun snapshot(rev: Int, hidden: Int) = JSONObject() - .put("type", "ADMIN_STATE").put("rev", rev).put("ts", 1_000L * rev).put("createdBy", me) + private fun snapshot(rev: Int, hidden: Int, by: String = me) = JSONObject() + .put("type", "ADMIN_STATE").put("rev", rev).put("ts", t0 + rev).put("createdBy", by) .put("state", JSONObject() .put("bannedMembers", JSONArray()) .put("hiddenMessageIds", JSONArray((0 until hidden).map { "message-id-%030d".format(it) })) @@ -75,11 +80,14 @@ class AdminStateSplitTest { .mapNotNull { it.optJSONObject("content") } .filter { it.optString("type") == "admin_invalidate" } - private fun open() { - manager.openChannel(streamId) + private fun open(channel: com.pombo.android.data.Channel = room) { + manager.openChannel(channel.messageStreamId) reads.clear() } + private fun signal(rev: Int) = h.deliver(theirs.ephemeralStreamId, StreamConstants.EPH_CONTROL, + JSONObject().put("type", "admin_invalidate").put("rev", rev).put("ts", t0 + rev), from = owner) + @Test fun `a snapshot that fits goes out whole, and rides the signal`() { storage += snapshot(1, 10) to me @@ -159,12 +167,11 @@ class AdminStateSplitTest { @Test fun `a signal without the snapshot makes a member read the -3`() { - storage += snapshot(1, 5) to me - open() - storage += SyncChunks.splitFramed(snapshot(2, 40), "r1", SyncChunks.ADMIN, 800).map { it to me } + storage += snapshot(1, 5, by = owner) to owner + open(theirs) + storage += SyncChunks.splitFramed(snapshot(2, 40, by = owner), "r1", SyncChunks.ADMIN, 800).map { it to owner } - h.deliver(room.ephemeralStreamId, StreamConstants.EPH_CONTROL, - JSONObject().put("type", "admin_invalidate").put("rev", 2).put("ts", 2_000L), from = me) + signal(2) assertTrue(reads.isNotEmpty()) assertEquals(40, manager.hiddenIds.value.size) @@ -214,23 +221,21 @@ class AdminStateSplitTest { @Test fun `a run that lands on the first read is not read again`() { - storage += snapshot(1, 5) to me - open() - storage += SyncChunks.splitFramed(snapshot(2, 40), "r1", SyncChunks.ADMIN, 800).map { it to me } + storage += snapshot(1, 5, by = owner) to owner + open(theirs) + storage += SyncChunks.splitFramed(snapshot(2, 40, by = owner), "r1", SyncChunks.ADMIN, 800).map { it to owner } - h.deliver(room.ephemeralStreamId, StreamConstants.EPH_CONTROL, - JSONObject().put("type", "admin_invalidate").put("rev", 2).put("ts", 2_000L), from = me) + signal(2) assertEquals(1, reads.size) } @Test fun `a run not on storage yet is read once more, and no more`() { - storage += snapshot(1, 5) to me - open() + storage += snapshot(1, 5, by = owner) to owner + open(theirs) - h.deliver(room.ephemeralStreamId, StreamConstants.EPH_CONTROL, - JSONObject().put("type", "admin_invalidate").put("rev", 2).put("ts", 2_000L), from = me) + signal(2) assertEquals(2, reads.size) assertEquals(5, manager.hiddenIds.value.size) @@ -238,11 +243,10 @@ class AdminStateSplitTest { @Test fun `a signal for a rev already held reads nothing`() { - storage += snapshot(3, 5) to me - open() + storage += snapshot(3, 5, by = owner) to owner + open(theirs) - h.deliver(room.ephemeralStreamId, StreamConstants.EPH_CONTROL, - JSONObject().put("type", "admin_invalidate").put("rev", 3).put("ts", 3_000L), from = me) + signal(3) assertTrue(reads.isEmpty()) } diff --git a/app/src/test/java/com/pombo/android/ChannelFieldStampsTest.kt b/app/src/test/java/com/pombo/android/ChannelFieldStampsTest.kt new file mode 100644 index 0000000..849ec61 --- /dev/null +++ b/app/src/test/java/com/pombo/android/ChannelFieldStampsTest.kt @@ -0,0 +1,88 @@ +package com.pombo.android + +import com.pombo.android.data.Channel +import io.mockk.every +import io.mockk.slot +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Test + +/** + * The sync merges a channel record field by field on these stamps, so a + * stamp on a field nobody changed here would let this device's stale value + * win it on every other device, and a lost stamp would let an older copy + * revert it. + */ +class ChannelFieldStampsTest { + + private val id = "0x00000000000000000000000000000000000000a1/c0ffee-1" + private val h = ChannelManagerHarness(listOf(ChannelManagerHarness.channel(id))) + private val saved = slot>() + + init { + every { h.store.save(capture(saved)) } returns Unit + } + + @After + fun tearDown() = h.stop() + + private fun stored() = saved.captured.single() + + @Test + fun `a save stamps only the fields it changes`() { + h.manager._channels.value = h.manager._channels.value.map { it.copy(name = "Renamed") } + + h.manager.saveChannels() + + assertEquals(setOf("name"), stored().fieldTs.keys) + assertEquals(stored(), h.manager._channels.value.single()) + } + + @Test + fun `a save that changes nothing stamps nothing`() { + h.manager.saveChannels() + + assertTrue(stored().fieldTs.isEmpty()) + } + + @Test + fun `what a sync pull brings in is not stamped, and its stamps are kept`() { + val imported = h.manager._channels.value.single() + .copy(name = "From another device", fieldTs = mapOf("name" to 5000L)) + h.manager.replaceChannels(listOf(imported)) + + h.manager.saveChannels() + + assertEquals(mapOf("name" to 5000L), stored().fieldTs) + } + + @Test + fun `a copy that lost its stamps does not take them back from the stored record`() { + val imported = h.manager._channels.value.single().copy(fieldTs = mapOf("name" to 5000L)) + h.manager.replaceChannels(listOf(imported)) + h.manager._channels.value = listOf(imported.copy(fieldTs = emptyMap(), description = "Edited")) + + h.manager.saveChannels() + + assertEquals(5000L, stored().fieldTs["name"]) + assertTrue("description" in stored().fieldTs) + } + + @Test + fun `a record created here carries no stamps`() { + val created = ChannelManagerHarness.channel("0x00000000000000000000000000000000000000a1/beef-1") + h.manager._channels.value = h.manager._channels.value + created + + h.manager.saveChannels() + + assertTrue(saved.captured.all { it.fieldTs.isEmpty() }) + } + + @Test + fun `the stamps survive the stored and synced form`() { + val channel = ChannelManagerHarness.channel(id).copy(fieldTs = mapOf("name" to 5000L, "storageDays" to 6000L)) + + assertEquals(channel, Channel.fromJson(channel.toJson())) + } +} diff --git a/app/src/test/java/com/pombo/android/ChannelManagerHarness.kt b/app/src/test/java/com/pombo/android/ChannelManagerHarness.kt index 88b9e6e..0ae8814 100644 --- a/app/src/test/java/com/pombo/android/ChannelManagerHarness.kt +++ b/app/src/test/java/com/pombo/android/ChannelManagerHarness.kt @@ -5,7 +5,9 @@ import com.pombo.android.data.Channel import io.mockk.coEvery import io.mockk.every import io.mockk.mockk +import io.mockk.mockkObject import io.mockk.slot +import io.mockk.unmockkObject import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob @@ -27,6 +29,9 @@ import org.json.JSONObject * The mocks are relaxed, so anything reached through the bridge answers with a * stub. Only assert on state this class wrote; a call that merely returned is * no evidence. + * + * The Graph is stubbed to know nothing. Its cache is JVM-wide: with real + * lookups, whether an open runs to the end depends on an earlier test. */ class ChannelManagerHarness( channels: List = emptyList(), @@ -60,6 +65,9 @@ class ChannelManagerHarness( every { ensStore.cachedName(any()) } returns null every { ensStore.cachedAvatar(any()) } returns null coEvery { ensStore.name(any(), any()) } returns null + mockkObject(com.pombo.android.core.GraphApi) + coEvery { com.pombo.android.core.GraphApi.streamRetention(any()) } returns null + coEvery { com.pombo.android.core.GraphApi.getChannelInfo(any()) } returns null val args = slot() coEvery { bridge.call(any(), capture(args)) } answers { @@ -101,7 +109,10 @@ class ChannelManagerHarness( ) } - fun stop() = scope.cancel() + fun stop() { + scope.cancel() + unmockkObject(com.pombo.android.core.GraphApi) + } companion object { fun channel( diff --git a/app/src/test/java/com/pombo/android/ChannelStaleCopyTest.kt b/app/src/test/java/com/pombo/android/ChannelStaleCopyTest.kt new file mode 100644 index 0000000..f553ea1 --- /dev/null +++ b/app/src/test/java/com/pombo/android/ChannelStaleCopyTest.kt @@ -0,0 +1,48 @@ +package com.pombo.android + +import io.mockk.coEvery +import kotlinx.coroutines.runBlocking +import org.json.JSONObject +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Test + +/** + * An edit that waits on the chain lands on the record as it is when the call + * returns. A save stamps every field that differs from the stored record, so + * an edit built on the copy opened before the call would revert, on every + * device, whatever a sync import brought in meanwhile. + */ +class ChannelStaleCopyTest { + + private val h = ChannelManagerHarness() + private val member = "0x00000000000000000000000000000000000000b1" + + @After + fun tearDown() = h.stop() + + @Test + fun `adding a member keeps what a sync import brought in while the grants ran`() { + val open = ChannelManagerHarness.channel("${h.me}/c0ffee-1") + h.manager.replaceChannels(listOf(open)) + h.manager._current.value = open + var imported = false + coEvery { h.bridge.call("setPermissions", any(), any()) } answers { + if (!imported) { + imported = true + h.manager.replaceChannels( + listOf(open.copy(name = "From another device", fieldTs = mapOf("name" to 5000L))) + ) + } + JSONObject() + } + + runBlocking { h.manager.addMember(member) } + + val stored = h.manager._channels.value.single() + assertEquals("From another device", stored.name) + assertEquals(5000L, stored.fieldTs["name"]) + assertEquals(listOf(member), stored.members) + assertEquals(stored, h.manager._current.value) + } +} diff --git a/app/src/test/java/com/pombo/android/core/SyncMergeVectorsTest.kt b/app/src/test/java/com/pombo/android/core/SyncMergeVectorsTest.kt index 357e960..87840c0 100644 --- a/app/src/test/java/com/pombo/android/core/SyncMergeVectorsTest.kt +++ b/app/src/test/java/com/pombo/android/core/SyncMergeVectorsTest.kt @@ -53,6 +53,23 @@ class SyncMergeVectorsTest { } } + @Test + fun `channel records merge field by field as on the web`() { + val cases = vectors().getJSONArray("channels") + assertTrue(cases.length() > 0) + for (i in 0 until cases.length()) { + val case = cases.getJSONObject(i) + val merged = SyncMerge.mergeState(case.getJSONObject("base"), case.getJSONObject("incoming")) + val expected = case.getJSONObject("expected") + for (key in listOf("channels", "channelsLeftAt")) { + assertTrue( + "${case.getString("what")}: $key was ${merged.opt(key)}, expected ${expected.opt(key)}", + same(merged.opt(key), expected.opt(key)) + ) + } + } + } + @Test fun `sent DMs carry deletions and edits as on the web`() { val cases = vectors().getJSONArray("sent") diff --git a/docs/SYNC-merge-vectors.json b/docs/SYNC-merge-vectors.json index 36e0382..1f8e342 100644 --- a/docs/SYNC-merge-vectors.json +++ b/docs/SYNC-merge-vectors.json @@ -835,5 +835,494 @@ } } } + ], + "channels": [ + { + "what": "a stamped field held here beats an old unstamped copy arriving", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "a stamped field arriving beats the old unstamped copy held here", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "two devices changing different fields keep both changes", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [ + "0x00000000000000000000000000000000000000b1" + ], + "fieldTs": { + "accessSnapshot": 6000 + } + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [ + "0x00000000000000000000000000000000000000b1" + ], + "fieldTs": { + "accessSnapshot": 6000, + "name": 5000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "the same two changes merged in the other order give the same record", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [ + "0x00000000000000000000000000000000000000b1" + ], + "fieldTs": { + "accessSnapshot": 6000 + } + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [ + "0x00000000000000000000000000000000000000b1" + ], + "fieldTs": { + "name": 5000, + "accessSnapshot": 6000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "a client that drops the stamps neither reverts a stamped field nor erases the stamps", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 90, + "accessSnapshot": [], + "fieldTs": { + "storageDays": 5000 + } + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [] + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 90, + "accessSnapshot": [], + "fieldTs": { + "storageDays": 5000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "an older stamp arriving does not undo a newer one held here", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Newer", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 6000 + } + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Older", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Newer", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 6000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "equal stamps fall back to the join-time rule, which gives the tie to the incoming copy", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Here", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Arriving", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Arriving", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 5000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "a field only one copy carries is kept, whichever copy wins the others", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "keysStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-4", + "storageProvider": "streamr" + } + ] + }, + "incoming": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "inboxStreamId": null, + "fieldTs": { + "name": 5000 + } + } + ] + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "inboxStreamId": null, + "keysStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-4", + "storageProvider": "streamr", + "fieldTs": { + "name": 5000 + } + } + ], + "channelsLeftAt": {} + } + }, + { + "what": "a leave newer than the join removes the channel whatever its stamps say", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Renamed", + "type": "public", + "createdAt": 1000, + "joinedAt": 1000, + "storageDays": 30, + "accessSnapshot": [], + "fieldTs": { + "name": 9000 + } + } + ] + }, + "incoming": { + "channels": [], + "channelsLeftAt": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": 5000 + } + }, + "expected": { + "channels": [], + "channelsLeftAt": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": 5000 + } + } + }, + { + "what": "a join newer than the leave keeps the channel and retires the leave", + "base": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 7000, + "storageDays": 30, + "accessSnapshot": [] + } + ] + }, + "incoming": { + "channels": [], + "channelsLeftAt": { + "0x00000000000000000000000000000000000000a1/c0ffee-1": 5000 + } + }, + "expected": { + "channels": [ + { + "messageStreamId": "0x00000000000000000000000000000000000000a1/c0ffee-1", + "name": "Channel", + "type": "public", + "createdAt": 1000, + "joinedAt": 7000, + "storageDays": 30, + "accessSnapshot": [] + } + ], + "channelsLeftAt": {} + } + } ] }