chore(model-server): use non-blocking method to send version delta

Use `respondBytesWriter` because it has a non-blocking `writeStringUtf8` method.
`respondTextWriter` has only blocking methods to write data.

Using a `java.io.Writer` is just an indirection without purpose.

Using `respondBytesWriter` needed to take a bug with Ktor in consideration. For the expected behavior, a test was added.

Author: Oleksandr Dzhychko

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

@@ -27,13 +27,17 @@ import io.ktor.server.resources.get
 import io.ktor.server.resources.post
 import io.ktor.server.resources.put
 import io.ktor.server.response.respond
+import io.ktor.server.response.respondBytesWriter
 import io.ktor.server.response.respondText
-import io.ktor.server.response.respondTextWriter
 import io.ktor.server.routing.Route
 import io.ktor.server.routing.route
 import io.ktor.server.routing.routing
 import io.ktor.server.websocket.webSocket
+import io.ktor.util.cio.use
 import io.ktor.util.pipeline.PipelineContext
+import io.ktor.utils.io.ByteWriteChannel
+import io.ktor.utils.io.close
+import io.ktor.utils.io.writeStringUtf8
 import io.ktor.websocket.send
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -84,6 +88,7 @@ class ModelReplicationServer(
     companion object {
         private val LOG = LoggerFactory.getLogger(ModelReplicationServer::class.java)
     }
+
     private val storeClient: IStoreClient get() = modelClient.store
 
     fun init(application: Application) {
@@ -378,24 +383,47 @@ class ModelReplicationServer(
         respond(delta)
     }
 
-    private suspend fun ApplicationCall.respondDeltaAsObjectStream(versionHash: String, baseVersionHash: String?, plainText: Boolean) {
+    private suspend fun ApplicationCall.respondDeltaAsObjectStream(
+        versionHash: String,
+        baseVersionHash: String?,
+        plainText: Boolean,
+    ) {
         // Call `computeDelta` before starting to respond.
         // It could already throw an exception, and in that case we do not want a successful response status.
         val objectData = repositoriesManager.computeDelta(versionHash, baseVersionHash)
-        respondTextWriter(contentType = if (plainText) ContentType.Text.Plain else VersionDeltaStream.CONTENT_TYPE) {
-            objectData.asFlow()
-                .flatten()
-                .withSeparator("\n")
-                .onEmpty { emit(versionHash) }
-                .withIndex()
-                .collect {
-                    if (it.index == 0) check(it.value == versionHash) { "First object should be the version" }
-                    append(it.value)
-                }
+        val contentType = if (plainText) ContentType.Text.Plain else VersionDeltaStream.CONTENT_TYPE
+        respondBytesWriter(contentType) {
+            this.useClosingWithoutCause {
+                objectData.asFlow()
+                    .flatten()
+                    .withSeparator("\n")
+                    .onEmpty { emit(versionHash) }
+                    .withIndex()
+                    .collect {
+                        if (it.index == 0) check(it.value == versionHash) { "First object should be the version" }
+                        writeStringUtf8(it.value)
+                    }
+            }
         }
     }
 }
 
+/**
+ * Same as [[ByteWriteChannel.use]] but closing without a cause in case of an exception.
+ *
+ * Calling [[ByteWriteChannel.close]] with a cause results in not closing the connection properly.
+ * See ModelReplicationServerTest.`server closes connection when failing to compute delta after starting to respond`
+ * This will only be fixed in Ktor 3.
+ * See https://youtrack.jetbrains.com/issue/KTOR-4862/Ktor-hangs-if-exception-occurs-during-write-response-body
+ */
+private inline fun ByteWriteChannel.useClosingWithoutCause(block: ByteWriteChannel.() -> Unit) {
+    try {
+        block()
+    } finally {
+        close()
+    }
+}
+
 private fun <T> Flow<Pair<T, T>>.flatten() = flow<T> {
     collect {
         emit(it.first)

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

@@ -50,6 +50,8 @@ import org.slf4j.LoggerFactory
 import java.lang.ref.SoftReference
 import java.util.UUID
 
+// The methods in this class are almost cohesive, so the number of functions is fine.
+@Suppress("complexity.TooManyFunctions")
 class RepositoriesManager(val client: LocalModelClient) : IRepositoriesManager {
 
     init {

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

@@ -19,11 +19,15 @@ package org.modelix.model.server
 import io.ktor.serialization.kotlinx.json.json
 import io.ktor.server.application.Application
 import io.ktor.server.application.install
+import io.ktor.server.netty.Netty
+import io.ktor.server.netty.NettyApplicationEngine
 import io.ktor.server.plugins.contentnegotiation.ContentNegotiation
 import io.ktor.server.resources.Resources
 import io.ktor.server.routing.IgnoreTrailingSlash
 import io.ktor.server.testing.ApplicationTestBuilder
 import io.ktor.server.websocket.WebSockets
+import kotlinx.coroutines.runBlocking
+import org.modelix.authorization.installAuthentication
 import org.modelix.model.client2.ModelClientV2
 
 suspend fun ApplicationTestBuilder.createModelClient(): ModelClientV2 {
@@ -37,3 +41,33 @@ fun Application.installDefaultServerPlugins() {
     install(Resources)
     install(IgnoreTrailingSlash)
 }
+
+/**
+ * Allow running a model server for test with Netty.
+ *
+ * This allows testing properties that would not be testable with [io.ktor.server.testing.testApplication]
+ * because the requests run on proper request threads.
+ *
+ * Examples for such properties are
+ * (1) errors while writing responses and
+ * (2) effects of blocking code on request threads.
+ */
+fun runWithNettyServer(
+    setupBlock: (application: Application) -> Unit,
+    testBlock: suspend (server: NettyApplicationEngine) -> Unit,
+) {
+    val nettyServer: NettyApplicationEngine = io.ktor.server.engine.embeddedServer(Netty, port = 0) {
+        installAuthentication(unitTestMode = true)
+        installDefaultServerPlugins()
+        setupBlock(this)
+    }
+
+    try {
+        nettyServer.start(wait = false)
+        runBlocking {
+            testBlock(nettyServer)
+        }
+    } finally {
+        nettyServer.stop()
+    }
+}

model-server/src/test/kotlin/org/modelix/model/server/handlers/ModelReplicationServerTest.kt

@@ -15,14 +15,25 @@
 
 package org.modelix.model.server.handlers
 
+import io.ktor.client.HttpClient
+import io.ktor.client.engine.cio.CIO
+import io.ktor.client.plugins.HttpTimeout
+import io.ktor.client.plugins.defaultRequest
 import io.ktor.client.request.get
 import io.ktor.http.HttpStatusCode
 import io.ktor.http.appendPathSegments
 import io.ktor.http.takeFrom
+import io.ktor.server.application.Application
+import io.ktor.server.netty.NettyApplicationEngine
 import io.ktor.server.testing.ApplicationTestBuilder
 import io.ktor.server.testing.testApplication
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.channelFlow
+import kotlinx.coroutines.flow.emitAll
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withTimeout
 import org.modelix.authorization.installAuthentication
 import org.modelix.model.InMemoryModels
 import org.modelix.model.api.IConceptReference
@@ -32,11 +43,13 @@ import org.modelix.model.client2.runWrite
 import org.modelix.model.client2.useVersionStreamFormat
 import org.modelix.model.lazy.RepositoryId
 import org.modelix.model.server.installDefaultServerPlugins
+import org.modelix.model.server.runWithNettyServer
 import org.modelix.model.server.store.InMemoryStoreClient
 import org.modelix.model.server.store.LocalModelClient
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.fail
+import kotlin.time.Duration.Companion.seconds
 
 class ModelReplicationServerTest {
 
@@ -47,7 +60,7 @@ class ModelReplicationServerTest {
         return ModelReplicationServer(repositoriesManager, modelClient, InMemoryModels())
     }
 
-    private fun runTest(
+    private fun runWithTestModelServer(
         modelReplicationServer: ModelReplicationServer = getDefaultModelReplicationServer(),
         block: suspend ApplicationTestBuilder.(scope: CoroutineScope) -> Unit,
     ) = testApplication {
@@ -63,7 +76,7 @@ class ModelReplicationServerTest {
     }
 
     @Test
-    fun `pulling delta does not return objects twice`() = runTest {
+    fun `pulling delta does not return objects twice`() = runWithTestModelServer {
         // Arrange
         val url = "http://localhost/v2"
         val modelClient = ModelClientV2.builder().url(url).client(client).build().also { it.init() }
@@ -117,7 +130,7 @@ class ModelReplicationServerTest {
         val repositoryId = RepositoryId("repo1")
         val branchRef = repositoryId.getBranchReference()
 
-        runTest(modelReplicationServer) {
+        runWithTestModelServer(modelReplicationServer) {
             repositoriesManager.createRepository(repositoryId, null)
 
             // Act
@@ -133,4 +146,63 @@ class ModelReplicationServerTest {
             assertEquals(HttpStatusCode.InternalServerError, response.status)
         }
     }
+
+    @Test
+    fun `server closes connection when failing to compute delta after starting to respond`() = runTest {
+        // Arrange
+        val repositoryId = RepositoryId("repo1")
+        val branchRef = repositoryId.getBranchReference()
+        val storeClient = InMemoryStoreClient()
+        val modelClient = LocalModelClient(storeClient)
+        val repositoriesManager = RepositoriesManager(modelClient)
+        val faultyRepositoriesManager = object :
+            IRepositoriesManager by repositoriesManager {
+            override suspend fun computeDelta(versionHash: String, baseVersionHash: String?): ObjectData {
+                val originalFlow = repositoriesManager.computeDelta(versionHash, baseVersionHash).asFlow()
+                val brokenFlow = channelFlow<Pair<String, String>> {
+                    error("Unexpected error.")
+                }
+                return ObjectDataFlow(
+                    flow {
+                        emitAll(originalFlow)
+                        emitAll(brokenFlow)
+                    },
+                )
+            }
+        }
+        repositoriesManager.createRepository(repositoryId, null)
+
+        suspend fun createClient(server: NettyApplicationEngine): HttpClient {
+            val port = server.resolvedConnectors().first().port
+            return HttpClient(CIO) {
+                defaultRequest {
+                    url("http://localhost:$port")
+                }
+                install(HttpTimeout) {
+                    requestTimeoutMillis = 5_000
+                }
+            }
+        }
+
+        val modelReplicationServer = ModelReplicationServer(faultyRepositoriesManager, modelClient, InMemoryModels())
+        val setupBlock = { application: Application -> modelReplicationServer.init(application) }
+        val testBlock: suspend (server: NettyApplicationEngine) -> Unit = { server ->
+            withTimeout(10.seconds) {
+                val client = createClient(server)
+                // Act
+                val response = client.get {
+                    url {
+                        takeFrom(url)
+                        appendPathSegments("v2", "repositories", repositoryId.id, "branches", branchRef.branchName)
+                    }
+                    useVersionStreamFormat()
+                }
+
+                // Assert
+                // The response should be delivered even if an exception is thrown inside the flow.
+                assertEquals(HttpStatusCode.OK, response.status)
+            }
+        }
+        runWithNettyServer(setupBlock, testBlock)
+    }
 }