Merge pull request #213 from modelix/MODELIX-515

MODELIX-515 ReplicatedRepository failed to sync with the server

Author: slisson

commitlint.config.js

@@ -13,6 +13,7 @@ module.exports = {
         "model-api-gen-gradle",
         "model-api",
         "model-client",
+        "model-datastructure",
         "model-server",
         "model-sync-lib",
         "modelql",

kotlin-js-store/yarn.lock

@@ -2,6 +2,39 @@
 # yarn lockfile v1
 
 
+"@aws-crypto/sha256-js@^5.0.0":
+  version "5.0.0"
+  resolved "https://registry.yarnpkg.com/@aws-crypto/sha256-js/-/sha256-js-5.0.0.tgz#fec6d5a9a097e812207eacaaa707bfa9191b3ad8"
+  integrity sha512-g+u9iKkaQVp9Mjoxq1IJSHj9NHGZF441+R/GIH0dn7u4mix5QQ4VqgpppHrNm1LzjUzb0BpcFGsBXP6cOVf+ZQ==
+  dependencies:
+    "@aws-crypto/util" "^5.0.0"
+    "@aws-sdk/types" "^3.222.0"
+    tslib "^1.11.1"
+
+"@aws-crypto/util@^5.0.0":
+  version "5.0.0"
+  resolved "https://registry.yarnpkg.com/@aws-crypto/util/-/util-5.0.0.tgz#afa286af897ea2bd9fab194b4a6be9cc562db23a"
+  integrity sha512-1GYqLdYRe96idcCltlqxdJ68OWE6ADT8qGLmVi7PVHKl8AxD2EWSbJSSevPq2eTx6vaPZpkr1RoZ3lcw/uGoEA==
+  dependencies:
+    "@aws-sdk/types" "^3.222.0"
+    "@aws-sdk/util-utf8-browser" "^3.0.0"
+    tslib "^1.11.1"
+
+"@aws-sdk/types@^3.222.0":
+  version "3.398.0"
+  resolved "https://registry.yarnpkg.com/@aws-sdk/types/-/types-3.398.0.tgz#8ce02559536670f9188cddfce32e9dd12b4fe965"
+  integrity sha512-r44fkS+vsEgKCuEuTV+TIk0t0m5ZlXHNjSDYEUvzLStbbfUFiNus/YG4UCa0wOk9R7VuQI67badsvvPeVPCGDQ==
+  dependencies:
+    "@smithy/types" "^2.2.2"
+    tslib "^2.5.0"
+
+"@aws-sdk/util-utf8-browser@^3.0.0":
+  version "3.259.0"
+  resolved "https://registry.yarnpkg.com/@aws-sdk/util-utf8-browser/-/util-utf8-browser-3.259.0.tgz#3275a6f5eb334f96ca76635b961d3c50259fd9ff"
+  integrity sha512-UvFa/vR+e19XookZF8RzFZBrw2EUkQWxiBW0yYQAhvk3C+QVGl0H3ouca8LDBlBfQKXwmW3huo/59H8rwb1wJw==
+  dependencies:
+    tslib "^2.3.1"
+
 "@colors/[email protected]":
   version "1.5.0"
   resolved "https://registry.yarnpkg.com/@colors/colors/-/colors-1.5.0.tgz#bb504579c1cae923e6576a4f5da43d25f97bdbd9"
@@ -78,6 +111,13 @@
 "@modelix/ts-model-api@file:../../ts-model-api":
   version "1.3.2-kernelf.4.dirty-SNAPSHOT"
 
+"@smithy/types@^2.2.2":
+  version "2.2.2"
+  resolved "https://registry.yarnpkg.com/@smithy/types/-/types-2.2.2.tgz#bd8691eb92dd07ac33b83e0e1c45f283502b1bf7"
+  integrity sha512-4PS0y1VxDnELGHGgBWlDksB2LJK8TG8lcvlWxIsgR+8vROI7Ms8h1P4FQUx+ftAX2QZv5g1CJCdhdRmQKyonyw==
+  dependencies:
+    tslib "^2.5.0"
+
 "@socket.io/component-emitter@~3.1.0":
   version "3.1.0"
   resolved "https://registry.yarnpkg.com/@socket.io/component-emitter/-/component-emitter-3.1.0.tgz#96116f2a912e0c02817345b3c10751069920d553"
@@ -2045,6 +2085,16 @@ tr46@~0.0.3:
   resolved "https://registry.yarnpkg.com/tr46/-/tr46-0.0.3.tgz#8184fd347dac9cdc185992f3a6622e14b9d9ab6a"
   integrity sha512-N3WMsuqV66lT30CrXNbEjx4GEwlow3v6rr4mCcv6prnfwhS01rkgyFdjPNBYd9br7LpXV1+Emh01fHnq2Gdgrw==
 
+tslib@^1.11.1:
+  version "1.14.1"
+  resolved "https://registry.yarnpkg.com/tslib/-/tslib-1.14.1.tgz#cf2d38bdc34a134bcaf1091c41f6619e2f672d00"
+  integrity sha512-Xni35NKzjgMrwevysHTCArtLDpPvye8zV/0E4EyYn43P7/7qvQwPh9BGkHewbMulVntbigmcT7rdX3BNo9wRJg==
+
+tslib@^2.3.1, tslib@^2.5.0:
+  version "2.6.2"
+  resolved "https://registry.yarnpkg.com/tslib/-/tslib-2.6.2.tgz#703ac29425e7b37cd6fd456e92404d46d1f3e4ae"
+  integrity sha512-AEYxH93jGFPn/a2iVAwW87VuUIkR1FVUKB77NwMF7nBTDkDrrT/Hpt/IrCJ0QXhW27jTBDcf5ZY7w6RiqTMw2Q==
+
 type-check@~0.3.2:
   version "0.3.2"
   resolved "https://registry.yarnpkg.com/type-check/-/type-check-0.3.2.tgz#5884cab512cf1d355e3fb784f30804b2b520db72"

model-client/build.gradle.kts

@@ -20,7 +20,13 @@ val kotlinxSerializationVersion: String by rootProject
 kotlin {
     jvm()
     js(IR) {
-        // browser {}
+        browser {
+            testTask {
+                useMocha {
+                    timeout = "30s"
+                }
+            }
+        }
         nodejs {
             testTask {
                 useMocha {
@@ -34,6 +40,7 @@ kotlin {
         val commonMain by getting {
             dependencies {
                 api(project(":model-api"))
+                api(project(":model-datastructure"))
                 api(project(":model-server-api"))
                 kotlin("stdlib-common")
                 implementation(libs.kotlin.collections.immutable)

model-client/src/commonMain/kotlin/org/modelix/model/client2/IModelClientV2.kt

@@ -34,8 +34,10 @@ interface IModelClientV2 {
     /**
      * The pushed version is merged automatically by the server with the current head.
      * The merge result is returned.
+     * @param baseVersion Some version that is known to exist on the server.
+     *                    Is used for optimizing the amount of data sent to the server.
      */
-    suspend fun push(branch: BranchReference, version: IVersion): IVersion
+    suspend fun push(branch: BranchReference, version: IVersion, baseVersion: IVersion?): IVersion
 
     suspend fun pull(branch: BranchReference, lastKnownVersion: IVersion?): IVersion
     suspend fun pull(branch: BranchReference, lastKnownVersion: IVersion?, filter: ModelQuery): IVersion

model-client/src/commonMain/kotlin/org/modelix/model/client2/ModelClientV2.kt

@@ -39,11 +39,11 @@ import org.modelix.model.lazy.BranchReference
 import org.modelix.model.lazy.CLVersion
 import org.modelix.model.lazy.ObjectStoreCache
 import org.modelix.model.lazy.RepositoryId
+import org.modelix.model.lazy.computeDelta
 import org.modelix.model.persistent.HashUtil
 import org.modelix.model.persistent.MapBaseStore
 import org.modelix.model.server.api.ModelQuery
 import org.modelix.model.server.api.v2.VersionDelta
-import kotlin.jvm.Synchronized
 import kotlin.time.Duration.Companion.seconds
 
 class ModelClientV2(
@@ -53,7 +53,7 @@ class ModelClientV2(
     private var clientId: Int = 0
     private var idGenerator: IIdGenerator = IdGeneratorDummy()
     private var userId: String? = null
-    private val kvStore = UncommitedEntriesStore()
+    private val kvStore = MapBaseStore()
     val store = ObjectStoreCache(kvStore) // TODO the store will accumulate garbage
 
     suspend fun init() {
@@ -129,17 +129,20 @@ class ModelClientV2(
         return createVersion(null, delta)
     }
 
-    override suspend fun push(branch: BranchReference, version: IVersion): IVersion {
+    override suspend fun push(branch: BranchReference, version: IVersion, baseVersion: IVersion?): IVersion {
+        LOG.debug { "${clientId.toString(16)}.push($branch, $version, $baseVersion)" }
         require(version is CLVersion)
+        require(baseVersion is CLVersion?)
         version.write()
-        val objects = kvStore.getUncommitedEntries().values.filterNotNull().toSet()
+        val objects = version.computeDelta(baseVersion)
         val response = httpClient.post {
             url {
                 takeFrom(baseUrl)
                 appendPathSegments("repositories", branch.repositoryId.id, "branches", branch.branchName)
             }
             contentType(ContentType.Application.Json)
-            val body = VersionDelta(version.hash, null, objects)
+            val body = VersionDelta(version.getContentHash(), null, objectsMap = objects)
+            body.checkObjectHashes()
             setBody(body)
         }
         val mergedVersionDelta = response.body<VersionDelta>()
@@ -157,7 +160,9 @@ class ModelClientV2(
                 }
             }
         }
-        return createVersion(lastKnownVersion, response.body())
+        val receivedVersion = createVersion(lastKnownVersion, response.body())
+        LOG.debug { "${clientId.toString(16)}.pull($branch, $lastKnownVersion) -> $receivedVersion" }
+        return receivedVersion
     }
 
     override suspend fun poll(branch: BranchReference, lastKnownVersion: IVersion?): IVersion {
@@ -171,7 +176,9 @@ class ModelClientV2(
                 }
             }
         }
-        return createVersion(lastKnownVersion, response.body())
+        val receivedVersion = createVersion(lastKnownVersion, response.body())
+        LOG.debug { "${clientId.toString(16)}.poll($branch, $lastKnownVersion) -> $receivedVersion" }
+        return receivedVersion
     }
 
     override suspend fun pull(branch: BranchReference, lastKnownVersion: IVersion?, filter: ModelQuery): IVersion {
@@ -186,13 +193,13 @@ class ModelClientV2(
         return if (baseVersion == null) {
             CLVersion(
                 delta.versionHash,
-                store.also { it.keyValueStore.putAll(delta.objects.associateBy { HashUtil.sha256(it) }) },
+                store.also { it.keyValueStore.putAll(delta.getAllObjects()) },
             )
         } else if (delta.versionHash == baseVersion.hash) {
             baseVersion
         } else {
             require(baseVersion.store == store) { "baseVersion was not created by this client" }
-            store.keyValueStore.putAll(delta.objects.associateBy { HashUtil.sha256(it) })
+            store.keyValueStore.putAll(delta.getAllObjects())
             CLVersion(
                 delta.versionHash,
                 baseVersion.store,
@@ -201,6 +208,7 @@ class ModelClientV2(
     }
 
     companion object {
+        private val LOG = mu.KotlinLogging.logger {}
         fun builder(): ModelClientV2Builder = ModelClientV2PlatformSpecificBuilder()
     }
 }
@@ -269,19 +277,8 @@ abstract class ModelClientV2Builder {
 
 expect class ModelClientV2PlatformSpecificBuilder() : ModelClientV2Builder
 
-private class UncommitedEntriesStore() : MapBaseStore() {
-    private var uncommitedEntries: MutableMap<String, String?> = HashMap()
-
-    @Synchronized
-    fun getUncommitedEntries(): Map<String, String?> {
-        val result = uncommitedEntries
-        uncommitedEntries = HashMap()
-        return result
-    }
-
-    @Synchronized
-    override fun putAll(entries: Map<String, String?>) {
-        uncommitedEntries.putAll(entries)
-        super.putAll(entries)
-    }
+fun VersionDelta.checkObjectHashes() {
+    HashUtil.checkObjectHashes(objectsMap)
 }
+
+fun VersionDelta.getAllObjects(): Map<String, String> = objectsMap + objects.associateBy { HashUtil.sha256(it) }

model-client/src/commonMain/kotlin/org/modelix/model/client2/ReplicatedModel.kt

@@ -3,6 +3,7 @@ package org.modelix.model.client2
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
@@ -50,13 +51,21 @@ class ReplicatedModel(val client: IModelClientV2, val branchRef: BranchReference
         otBranch = OTBranch(rawBranch, client.getIdGenerator(), lastRemoteVersion.store)
 
         scope.launch {
+            var nextDelayMs: Long = 0
             while (state != State.Disposed) {
-                val newRemoteVersion = if (query == null) {
-                    client.poll(branchRef, lastRemoteVersion)
-                } else {
-                    client.poll(branchRef, lastRemoteVersion, query)
-                } as CLVersion
-                remoteVersionReceived(newRemoteVersion)
+                if (nextDelayMs > 0) delay(nextDelayMs)
+                try {
+                    val newRemoteVersion = if (query == null) {
+                        client.poll(branchRef, lastRemoteVersion)
+                    } else {
+                        client.poll(branchRef, lastRemoteVersion, query)
+                    } as CLVersion
+                    remoteVersionReceived(newRemoteVersion)
+                    nextDelayMs = 0
+                } catch (ex: Throwable) {
+                    LOG.error(ex) { "Failed to poll branch $branchRef" }
+                    nextDelayMs = (nextDelayMs * 3 / 2).coerceIn(1000, 30000)
+                }
             }
         }
 
@@ -111,7 +120,7 @@ class ReplicatedModel(val client: IModelClientV2, val branchRef: BranchReference
             createdVersion = applyPendingLocalChanges()
         }
         if (createdVersion != null) {
-            remoteVersionReceived(client.push(branchRef, createdVersion) as CLVersion)
+            remoteVersionReceived(client.push(branchRef, createdVersion, baseVersion = lastRemoteVersion) as CLVersion)
         }
     }
 
@@ -147,6 +156,11 @@ class ReplicatedModel(val client: IModelClientV2, val branchRef: BranchReference
     }
 }
 
-fun IModelClientV2.getReplicatedModel(branchRef: BranchReference, query: ModelQuery? = null): ReplicatedModel {
+fun IModelClientV2.getReplicatedModel(branchRef: BranchReference): ReplicatedModel {
+    return ReplicatedModel(this, branchRef)
+}
+
+@Deprecated("ModelQuery is not supported and ignored", ReplaceWith("getReplicatedModel(branchRef)"))
+fun IModelClientV2.getReplicatedModel(branchRef: BranchReference, query: ModelQuery?): ReplicatedModel {
     return ReplicatedModel(this, branchRef, query)
 }

model-client/src/commonMain/kotlin/org/modelix/model/persistent/PlatformSpecificHashUtil.kt

@@ -22,6 +22,16 @@ class SerializationUtilEscapeTest {
         assertEquals("", SerializationUtil.escape(""))
     }
 
+    @Test
+    fun escape_space() {
+        assertEquals("+", SerializationUtil.escape(" "))
+    }
+
+    @Test
+    fun unescape_space() {
+        assertEquals(" ", SerializationUtil.unescape("+"))
+    }
+
     @Test
     fun unescape_emptyString() {
         assertEquals("", SerializationUtil.unescape(""))