fix(model-server): respond with error if delta computation fails before creating a flow

Before this fix, the server started responding with status code 200 before even trying to execute `repositoriesManager.computeDelta`.

Author: Oleksandr Dzhychko

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

@@ -379,8 +379,11 @@ class ModelReplicationServer(
     }
 
     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) {
-            repositoriesManager.computeDelta(versionHash, baseVersionHash).asFlow()
+            objectData.asFlow()
                 .flatten()
                 .withSeparator("\n")
                 .onEmpty { emit(versionHash) }

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

@@ -16,6 +16,7 @@
 package org.modelix.model.server.handlers
 
 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.testing.ApplicationTestBuilder
@@ -34,20 +35,26 @@ import org.modelix.model.server.installDefaultServerPlugins
 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
 
 class ModelReplicationServerTest {
 
-    private fun runTest(block: suspend ApplicationTestBuilder.(scope: CoroutineScope) -> Unit) = testApplication {
+    private fun getDefaultModelReplicationServer(): ModelReplicationServer {
+        val storeClient = InMemoryStoreClient()
+        val modelClient = LocalModelClient(storeClient)
+        val repositoriesManager = RepositoriesManager(modelClient)
+        return ModelReplicationServer(repositoriesManager, modelClient, InMemoryModels())
+    }
+
+    private fun runTest(
+        modelReplicationServer: ModelReplicationServer = getDefaultModelReplicationServer(),
+        block: suspend ApplicationTestBuilder.(scope: CoroutineScope) -> Unit,
+    ) = testApplication {
         application {
             installAuthentication(unitTestMode = true)
             installDefaultServerPlugins()
-            val storeClient = InMemoryStoreClient()
-            val modelClient = LocalModelClient(storeClient)
-            val repositoriesManager = RepositoriesManager(modelClient)
-            val inMemoryModels = InMemoryModels()
-            ModelReplicationServer(repositoriesManager, modelClient, inMemoryModels).init(this)
-            KeyValueLikeModelServer(repositoriesManager, storeClient, inMemoryModels).init(this)
+            modelReplicationServer.init(this)
         }
 
         coroutineScope {
@@ -92,4 +99,38 @@ class ModelReplicationServerTest {
             }
         }
     }
+
+    @Test
+    fun `server responds with error when failing to compute delta before starting to respond`() {
+        // Arrange
+        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 {
+                error("Unexpected error.")
+            }
+        }
+        val modelReplicationServer = ModelReplicationServer(faultyRepositoriesManager, modelClient, InMemoryModels())
+        val url = "http://localhost/v2"
+        val repositoryId = RepositoryId("repo1")
+        val branchRef = repositoryId.getBranchReference()
+
+        runTest(modelReplicationServer) {
+            repositoriesManager.createRepository(repositoryId, null)
+
+            // Act
+            val response = client.get {
+                url {
+                    takeFrom(url)
+                    appendPathSegments("repositories", repositoryId.id, "branches", branchRef.branchName)
+                }
+                useVersionStreamFormat()
+            }
+
+            // Assert
+            assertEquals(HttpStatusCode.InternalServerError, response.status)
+        }
+    }
 }