Merge pull request #534 from modelix/fix/send-all-objects-when-channel-becomes-full

fix(model-server): RepositoriesManager.computeDelta should wait for channel to have capacity

Author: Oleksandr Dzhychko

model-server/src/main/kotlin/org/modelix/model/server/handlers/RepositoriesManager.kt

@@ -18,6 +18,7 @@ import kotlinx.coroutines.flow.asFlow
 import kotlinx.coroutines.flow.channelFlow
 import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.runBlocking
 import kotlinx.datetime.Clock
 import org.apache.commons.collections4.map.LRUMap
 import org.modelix.model.IKeyValueStore
@@ -30,7 +31,6 @@ import org.modelix.model.api.IdGeneratorDummy
 import org.modelix.model.api.PBranch
 import org.modelix.model.api.runSynchronized
 import org.modelix.model.lazy.BranchReference
-import org.modelix.model.lazy.BulkQuery
 import org.modelix.model.lazy.CLTree
 import org.modelix.model.lazy.CLVersion
 import org.modelix.model.lazy.IDeserializingKeyValueStore
@@ -46,6 +46,7 @@ import java.lang.ref.SoftReference
 import java.util.UUID
 
 class RepositoriesManager(val client: LocalModelClient) {
+
     init {
         migrateLegacyRepositoriesList()
     }
@@ -246,19 +247,23 @@ class RepositoriesManager(val client: LocalModelClient) {
                 // we have to do this to not emit objects more than once.
                 val seenHashes = mutableSetOf<String>()
                 fun emitObjects(entry: KVEntryReference<*>) {
+                    if (seenHashes.contains(entry.getHash())) return
+                    seenHashes.add(entry.getHash())
                     bulkQuery.get(entry).onSuccess {
-                        channel.trySend(entry.getHash() to it!!.serialize())
-                        for (referencedEntry in it.getReferencedEntries()) {
-                            val wasSeenBefore = !seenHashes.add(referencedEntry.getHash())
-                            // Do not emit the object if we already emitted it.
-                            if (!wasSeenBefore) {
-                                emitObjects(referencedEntry)
-                            }
+                        val value = checkNotNull(it) { "No value received for ${entry.getHash()}" }
+                        // Use `send` instead of `trySend`,
+                        // because `trySend` fails if the channel capacity is full.
+                        // This might happen if the data is produced faster than consumed.
+                        // A better solution would be to have bulk queries which itself are asynchronous
+                        // but doing that needs more consideration.
+                        runBlocking { channel.send(entry.getHash() to value.serialize()) }
+                        for (referencedEntry in value.getReferencedEntries()) {
+                            emitObjects(referencedEntry)
                         }
                     }
                 }
                 emitObjects(KVEntryReference(versionHash, CPVersion.DESERIALIZER))
-                (bulkQuery as? BulkQuery)?.process()
+                bulkQuery.process()
             }
         }
 

model-server/src/test/kotlin/org/modelix/model/server/ModelClientV2Test.kt

@@ -196,7 +196,10 @@ class ModelClientV2Test {
         val repositoryId = RepositoryId("repo1")
         val branchId = repositoryId.getBranchReference("my-branch")
         modelClientForArrange.runWrite(branchId) { root ->
-            root.addNewChild("aChild", -1, null as IConceptReference?)
+            // Creating many children makes the flow emitting many values at once.
+            repeat(100) {
+                root.addNewChild("aChild", -1, null as IConceptReference?)
+            }
         }
 
         // Act