perf(model-client): send large changes as multiple HTTP requests

Author: slisson

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

@@ -23,6 +23,7 @@ import io.ktor.client.plugins.contentnegotiation.ContentNegotiation
 import io.ktor.client.plugins.expectSuccess
 import io.ktor.client.request.get
 import io.ktor.client.request.post
+import io.ktor.client.request.put
 import io.ktor.client.request.setBody
 import io.ktor.client.statement.bodyAsText
 import io.ktor.http.ContentType
@@ -178,20 +179,40 @@ class ModelClientV2(
         require(baseVersion is CLVersion?)
         version.write()
         val objects = version.computeDelta(baseVersion)
+        HashUtil.checkObjectHashes(objects)
+        val delta = if (objects.size > 1000) {
+            // large HTTP requests and large Json objects don't scale well
+            uploadObjects(branch.repositoryId, objects.asSequence().map { it.key to it.value })
+            VersionDelta(version.getContentHash(), null)
+        } else {
+            VersionDelta(version.getContentHash(), null, objectsMap = objects)
+        }
         val response = httpClient.post {
             url {
                 takeFrom(baseUrl)
                 appendPathSegmentsEncodingSlash("repositories", branch.repositoryId.id, "branches", branch.branchName)
             }
             contentType(ContentType.Application.Json)
-            val body = VersionDelta(version.getContentHash(), null, objectsMap = objects)
-            body.checkObjectHashes()
-            setBody(body)
+            setBody(delta)
         }
         val mergedVersionDelta = response.body<VersionDelta>()
         return createVersion(version, mergedVersionDelta)
     }
 
+    private suspend fun uploadObjects(repository: RepositoryId, objects: Sequence<Pair<String, String>>) {
+        LOG.debug { "${clientId.toString(16)}.pushObjects($repository)" }
+        objects.chunked(100_000).forEach { chunk ->
+            httpClient.put {
+                url {
+                    takeFrom(baseUrl)
+                    appendPathSegmentsEncodingSlash("repositories", repository.id, "objects")
+                }
+                contentType(ContentType.Text.Plain)
+                setBody(chunk.flatMap { it.toList() }.joinToString("\n"))
+            }
+        }
+    }
+
     override suspend fun pull(branch: BranchReference, lastKnownVersion: IVersion?): IVersion {
         require(lastKnownVersion is CLVersion?)
         val response = httpClient.get {

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

@@ -20,17 +20,21 @@ import io.ktor.server.application.ApplicationCall
 import io.ktor.server.application.call
 import io.ktor.server.plugins.origin
 import io.ktor.server.request.receive
+import io.ktor.server.request.receiveStream
 import io.ktor.server.response.respond
 import io.ktor.server.response.respondText
 import io.ktor.server.routing.Route
 import io.ktor.server.routing.get
 import io.ktor.server.routing.post
+import io.ktor.server.routing.put
 import io.ktor.server.routing.route
 import io.ktor.server.routing.routing
 import io.ktor.server.websocket.webSocket
 import io.ktor.util.pipeline.PipelineContext
 import io.ktor.websocket.send
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
+import kotlinx.coroutines.withContext
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
 import org.modelix.authorization.getUserName
@@ -44,6 +48,7 @@ import org.modelix.model.lazy.CLTree
 import org.modelix.model.lazy.CLVersion
 import org.modelix.model.lazy.RepositoryId
 import org.modelix.model.operations.OTBranch
+import org.modelix.model.persistent.HashUtil
 import org.modelix.model.server.api.v2.VersionDelta
 import org.modelix.model.server.store.IStoreClient
 import org.modelix.model.server.store.LocalModelClient
@@ -211,6 +216,32 @@ class ModelReplicationServer(val repositoriesManager: RepositoriesManager) {
                         }
                     }
                 }
+                route("objects") {
+                    put {
+                        var writtenEntries = 0
+                        withContext(Dispatchers.IO) {
+                            var isKey = true
+                            var key = ""
+                            call.receiveStream().bufferedReader().lineSequence().forEach { line ->
+                                if (isKey) {
+                                    key = line
+                                } else {
+                                    val value = line
+                                    require(HashUtil.isSha256(key)) {
+                                        "This API cannot be used to store other entries than serialized objects." +
+                                            " The key is expected to be a SHA256 hash over the value: $key -> $value"
+                                    }
+                                    val expectedKey = HashUtil.sha256(value)
+                                    require(expectedKey == key) { "Hash mismatch. Expected $expectedKey, but $key was provided. Value: $value" }
+                                    storeClient.put(key, value, true)
+                                    writtenEntries++
+                                }
+                                isKey = !isKey
+                            }
+                        }
+                        call.respondText("$writtenEntries objects received")
+                    }
+                }
             }
         }
         route("versions") {