prevent zombie presence events

yorkie-team · Jun 17, 2024 · 930f076 · 930f076
1 parent 1367862
commit 930f076
Show file tree

Hide file tree

Showing 4 changed files with 75 additions and 19 deletions.
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -117,9 +117,10 @@ jobs:
       - uses: actions/download-artifact@v2
       - uses: codecov/codecov-action@v4
         with:
-          token: ${{ secrets.CODECOV_TOKEN }}
           fail_ci_if_error: true
           verbose: false
+        env:
+          CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
 
   microbenchmarks:
     name: Microbenchmarks

diff --git a/yorkie/src/androidTest/kotlin/dev/yorkie/core/PresenceTest.kt b/yorkie/src/androidTest/kotlin/dev/yorkie/core/PresenceTest.kt
@@ -827,5 +827,51 @@ class PresenceTest {
         }
     }
 
+    @Test
+    fun test_whether_presence_event_queue_is_empty_after_consecutive_presence_changes() {
+        withTwoClientsAndDocuments { c1, _, d1, d2, _ ->
+            val d1PresenceEvents = mutableListOf<MyPresence.PresenceChanged>()
+            val d2PresenceEvents = mutableListOf<Others.PresenceChanged>()
+            val jobs = listOf(
+                launch(start = CoroutineStart.UNDISPATCHED) {
+                    d1.events.filterIsInstance<MyPresence.PresenceChanged>()
+                        .collect(d1PresenceEvents::add)
+                },
+                launch(start = CoroutineStart.UNDISPATCHED) {
+                    d2.events.filterIsInstance<Others.PresenceChanged>()
+                        .collect(d2PresenceEvents::add)
+                },
+            )
+
+            d1.updateAsync { _, presence ->
+                repeat(10) {
+                    presence.put(mapOf("a" to "${it + 1}"))
+                }
+            }.await()
+
+            val lastD1PresenceEvent = MyPresence.PresenceChanged(
+                PresenceInfo(c1.requireClientId(), mapOf("a" to "10")),
+            )
+            val lastD2PresenceEvent = Others.PresenceChanged(
+                PresenceInfo(c1.requireClientId(), mapOf("a" to "10")),
+            )
+
+            withTimeout(GENERAL_TIMEOUT) {
+                while (lastD1PresenceEvent !in d1PresenceEvents ||
+                    lastD2PresenceEvent !in d2PresenceEvents
+                ) {
+                    delay(50)
+                }
+            }
+
+            assertEquals(lastD1PresenceEvent, d1PresenceEvents.last())
+            assertTrue(d1.presenceEventQueue.isEmpty())
+            assertEquals(lastD2PresenceEvent, d2PresenceEvents.last())
+            assertTrue(d2.presenceEventQueue.isEmpty())
+
+            jobs.forEach(Job::cancel)
+        }
+    }
+
     private data class Cursor(val x: Int, val y: Int)
 }
diff --git a/yorkie/src/androidTest/kotlin/dev/yorkie/document/json/JsonTreeSplitMergeTest.kt b/yorkie/src/androidTest/kotlin/dev/yorkie/document/json/JsonTreeSplitMergeTest.kt
@@ -5,6 +5,7 @@ import dev.yorkie.TreeTest
 import dev.yorkie.core.Client.SyncMode.Manual
 import dev.yorkie.core.withTwoClientsAndDocuments
 import dev.yorkie.document.json.JsonTreeTest.Companion.rootTree
+import dev.yorkie.document.json.JsonTreeTest.Companion.updateAndSync
 import kotlin.test.assertEquals
 import org.junit.Test
 import org.junit.runner.RunWith
@@ -16,7 +17,7 @@ class JsonTreeSplitMergeTest {
     @Test
     fun test_contained_split_and_split_at_the_same_position() {
         withTwoClientsAndDocuments(syncMode = Manual) { c1, c2, d1, d2, _ ->
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.setNewTree(
                         "t",
@@ -31,7 +32,7 @@ class JsonTreeSplitMergeTest {
             )
             JsonTreeTest.assertTreesXmlEquals("<r><p>ab</p></r>", d1, d2)
 
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.rootTree().edit(2, 2, 1)
                 },
@@ -48,7 +49,7 @@ class JsonTreeSplitMergeTest {
     @Test
     fun test_contained_split_and_split_at_different_positions_on_the_same_node() {
         withTwoClientsAndDocuments(syncMode = Manual) { c1, c2, d1, d2, _ ->
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.setNewTree(
                         "t",
@@ -63,7 +64,7 @@ class JsonTreeSplitMergeTest {
             )
             JsonTreeTest.assertTreesXmlEquals("<r><p>abc</p></r>", d1, d2)
 
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.rootTree().edit(2, 2, 1)
                 },
@@ -81,7 +82,7 @@ class JsonTreeSplitMergeTest {
     @Test
     fun test_contained_split_and_insert_into_the_split_position() {
         withTwoClientsAndDocuments(syncMode = Manual) { c1, c2, d1, d2, _ ->
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.setNewTree(
                         "t",
@@ -96,7 +97,7 @@ class JsonTreeSplitMergeTest {
             )
             JsonTreeTest.assertTreesXmlEquals("<r><p>ab</p></r>", d1, d2)
 
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.rootTree().edit(2, 2, 1)
                 },
@@ -114,7 +115,7 @@ class JsonTreeSplitMergeTest {
     @Test
     fun test_contained_split_and_insert_into_original_node() {
         withTwoClientsAndDocuments(syncMode = Manual) { c1, c2, d1, d2, _ ->
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.setNewTree(
                         "t",
@@ -129,7 +130,7 @@ class JsonTreeSplitMergeTest {
             )
             JsonTreeTest.assertTreesXmlEquals("<r><p>ab</p></r>", d1, d2)
 
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.rootTree().edit(2, 2, 1)
                 },
@@ -147,7 +148,7 @@ class JsonTreeSplitMergeTest {
     @Test
     fun test_contained_split_and_insert_into_split_node() {
         withTwoClientsAndDocuments(syncMode = Manual) { c1, c2, d1, d2, _ ->
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.setNewTree(
                         "t",
@@ -162,7 +163,7 @@ class JsonTreeSplitMergeTest {
             )
             JsonTreeTest.assertTreesXmlEquals("<r><p>ab</p></r>", d1, d2)
 
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.rootTree().edit(2, 2, 1)
                 },
@@ -180,7 +181,7 @@ class JsonTreeSplitMergeTest {
     @Test
     fun test_contained_split_and_delete_contents_in_split_node() {
         withTwoClientsAndDocuments(syncMode = Manual) { c1, c2, d1, d2, _ ->
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.setNewTree(
                         "t",
@@ -195,7 +196,7 @@ class JsonTreeSplitMergeTest {
             )
             JsonTreeTest.assertTreesXmlEquals("<r><p>ab</p></r>", d1, d2)
 
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.rootTree().edit(2, 2, 1)
                 },
@@ -232,7 +233,7 @@ class JsonTreeSplitMergeTest {
             }.await()
             JsonTreeTest.assertTreesXmlEquals("<doc><p></p><p>ab</p></doc>", d1)
 
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.rootTree().edit(1, 3)
                 },
@@ -268,7 +269,7 @@ class JsonTreeSplitMergeTest {
                 d1.getRoot().rootTree().toXml(),
             )
 
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.rootTree().edit(2, 6)
                 },
@@ -309,7 +310,7 @@ class JsonTreeSplitMergeTest {
             }.await()
             assertEquals("<doc><p>a</p><p>c</p><p>b</p></doc>", d1.getRoot().rootTree().toXml())
 
-            JsonTreeTest.updateAndSync(
+            updateAndSync(
                 JsonTreeTest.Companion.Updater(c1, d1) { root, _ ->
                     root.rootTree().edit(2, 7)
                 },

diff --git a/yorkie/src/main/kotlin/dev/yorkie/document/Document.kt b/yorkie/src/main/kotlin/dev/yorkie/document/Document.kt
@@ -100,7 +100,8 @@ public class Document(
     public val garbageLength: Int
         get() = root.garbageLength
 
-    private val presenceEventQueue = mutableListOf<PresenceChanged>()
+    @VisibleForTesting
+    internal val presenceEventQueue = mutableListOf<PresenceChanged>()
     private val pendingPresenceEvents = mutableListOf<PresenceChanged>()
 
     private val onlineClients = MutableStateFlow(setOf<ActorID>())
@@ -360,6 +361,7 @@ public class Document(
      */
     private suspend fun publishPresenceEvent(presences: Presences) {
         val iterator = presenceEventQueue.listIterator()
+        var clearPresenceEventQueue = false
         while (iterator.hasNext()) {
             val event = iterator.next()
             if (event is Others && event.changed.actorID == changeID.actor) {
@@ -368,10 +370,16 @@ public class Document(
             }
 
             if (presenceEventReadyToBePublished(event, presences)) {
+                if (presenceEventQueue.first() != event) {
+                    clearPresenceEventQueue = true
+                }
                 eventStream.emit(event)
                 iterator.remove()
             }
         }
+        if (clearPresenceEventQueue) {
+            presenceEventQueue.clear()
+        }
     }
 
     private fun presenceEventReadyToBePublished(
@@ -382,14 +390,14 @@ public class Document(
             is MyPresence.Initialized -> presences.keys.containsAll(event.initialized.keys)
             is MyPresence.PresenceChanged -> {
                 val actorID = event.changed.actorID
-                actorID !in presences || event.changed.presence == presences[actorID]
+                event.changed.presence == presences[actorID]
             }
 
             is Others.Watched -> event.changed.actorID in presences
             is Others.Unwatched -> event.changed.actorID !in presences
             is Others.PresenceChanged -> {
                 val actorID = event.changed.actorID
-                actorID !in presences || event.changed.presence == presences[actorID]
+                event.changed.presence == presences[actorID]
             }
         }
     }