fix(model-client): detect incomplete version delta streams

Use a new version of the delta stream format that enables the client to properly detect incomplete delta version streams.

Author: Oleksandr Dzhychko

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

@@ -65,6 +65,7 @@ import org.modelix.model.persistent.HashUtil
 import org.modelix.model.persistent.MapBasedStore
 import org.modelix.model.server.api.v2.VersionDelta
 import org.modelix.model.server.api.v2.VersionDeltaStream
+import org.modelix.model.server.api.v2.VersionDeltaStreamV2
 import org.modelix.model.server.api.v2.asStream
 import org.modelix.modelql.client.ModelQLClient
 import org.modelix.modelql.core.IMonoStep
@@ -491,32 +492,47 @@ private fun URLBuilder.appendPathSegmentsEncodingSlash(vararg components: String
 fun VersionDelta.getAllObjects(): Map<String, String> = objectsMap + objects.associateBy { HashUtil.sha256(it) }
 
 suspend fun HttpResponse.readVersionDelta(): VersionDeltaStream {
-    return if (contentType()?.match(VersionDeltaStream.CONTENT_TYPE) == true) {
-        val content = bodyAsChannel()
-        val versionHash = checkNotNull(content.readUTF8Line()) { "No objects received" }
-        val versionObject = content.readUTF8Line()
-        return if (versionObject == null) {
-            VersionDeltaStream(versionHash, emptyFlow())
-        } else {
-            VersionDeltaStream(
-                versionHash,
-                flow {
-                    emit(versionHash to versionObject)
-                    while (true) {
-                        val key = content.readUTF8Line() ?: break
-                        val value = checkNotNull(content.readUTF8Line()) { "Object missing for hash $key" }
-                        emit(key to value)
-                    }
-                },
-            )
-        }
+    val parsedContentType = contentType()
+    return if (parsedContentType?.match(VersionDeltaStreamV2.CONTENT_TYPE) == true) {
+        return readVersionDeltaStreamV2()
+    } else if (parsedContentType?.match(VersionDeltaStream.CONTENT_TYPE) == true) {
+        return readVersionDeltaStreamV1()
     } else {
         body<VersionDelta>().asStream()
     }
 }
 
+private suspend fun HttpResponse.readVersionDeltaStreamV1(): VersionDeltaStream {
+    val content = bodyAsChannel()
+    val versionHash = checkNotNull(content.readUTF8Line()) { "No objects received" }
+    val versionObject = content.readUTF8Line()
+    return if (versionObject == null) {
+        VersionDeltaStream(versionHash, emptyFlow())
+    } else {
+        VersionDeltaStream(
+            versionHash,
+            flow {
+                emit(versionHash to versionObject)
+                while (true) {
+                    val key = content.readUTF8Line() ?: break
+                    val value = checkNotNull(content.readUTF8Line()) { "Object missing for hash $key" }
+                    emit(key to value)
+                }
+            },
+        )
+    }
+}
+
+private suspend fun HttpResponse.readVersionDeltaStreamV2(): VersionDeltaStream {
+    val content = bodyAsChannel()
+    val decodeVersionDeltaStreamV2 = VersionDeltaStreamV2.decodeVersionDeltaStreamV2(content)
+    return VersionDeltaStream(decodeVersionDeltaStreamV2.versionHash, decodeVersionDeltaStreamV2.hashesWithDeltaObject)
+}
+
 fun HttpRequestBuilder.useVersionStreamFormat() {
-    headers.set(HttpHeaders.Accept, VersionDeltaStream.CONTENT_TYPE.toString())
+    headers[HttpHeaders.Accept] = VersionDeltaStreamV2.CONTENT_TYPE.toString()
+    // Add CONTENT_TYPE_VERSION_DELTA_V1 so that newer clients cant talk with older servers.
+    headers.append(HttpHeaders.Accept, VersionDeltaStream.CONTENT_TYPE.toString())
 }
 
 /**

model-client/src/commonTest/kotlin/org/modelix/model/client2/ModelClientV2Test.kt

@@ -17,11 +17,17 @@ package org.modelix.model.client2
 
 import io.ktor.client.HttpClient
 import io.ktor.client.engine.mock.MockEngine
+import io.ktor.client.engine.mock.respond
 import io.ktor.client.engine.mock.respondError
+import io.ktor.http.HttpHeaders
 import io.ktor.http.HttpStatusCode
+import io.ktor.http.headersOf
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.test.runTest
+import org.modelix.model.lazy.RepositoryId
+import org.modelix.model.server.api.v2.VersionDeltaStreamV2
 import kotlin.test.Test
+import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
 
 class ModelClientV2Test {
@@ -38,8 +44,30 @@ class ModelClientV2Test {
             .url(url)
             .build()
         modelClient.close()
-        assertFailsWith<CancellationException>("Parent job is Completed") {
+        val exception = assertFailsWith<CancellationException> {
             modelClient.init()
         }
+        assertEquals("Parent job is Completed", exception.message)
+    }
+
+    @Test
+    fun detectIncompleteDataInVersionDeltaStreamV2() = runTest {
+        val incompleteData = "CTVRw*a6KXJ4o7uzGlp-kUosxpyRf4f"
+        val repositoryId = RepositoryId("aRepositoryId")
+        val branchRef = repositoryId.getBranchReference("main")
+        val url = "http://localhost/v2"
+        val mockEngine = MockEngine { requestData ->
+            assertEquals(VersionDeltaStreamV2.CONTENT_TYPE.toString(), requestData.headers[HttpHeaders.Accept])
+            respond(incompleteData, HttpStatusCode.OK, headersOf(HttpHeaders.ContentType, VersionDeltaStreamV2.CONTENT_TYPE.toString()))
+        }
+        val httpClient = HttpClient(mockEngine)
+        val modelClient = ModelClientV2.builder()
+            .client(httpClient)
+            .url(url)
+            .build()
+
+        assertFailsWith<VersionDeltaStreamV2.Companion.IncompleteData> {
+            modelClient.pull(branchRef, null)
+        }
     }
 }

model-client/src/jvmTest/kotlin/org/modelix/model/client2/ModelClientV2JvmTest.kt

@@ -22,6 +22,7 @@ import io.ktor.http.HttpStatusCode
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.test.runTest
 import kotlin.test.Test
+import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
 
 class ModelClientV2JvmTest {
@@ -40,13 +41,15 @@ class ModelClientV2JvmTest {
         // Implementing `close` allow to use `.use` method.
         modelClient.use {
         }
-        assertFailsWith<CancellationException>("Parent job is Completed") {
+        val firstException = assertFailsWith<CancellationException> {
             modelClient.init()
         }
-        // `Closable` implies, that `.close` method is idempotent.
+        assertEquals("Parent job is Completed", firstException.message)
+        // `Closable` implies that `.close` method is idempotent.
         modelClient.close()
-        assertFailsWith<CancellationException>("Parent job is Completed") {
+        val secondException = assertFailsWith<CancellationException> {
             modelClient.init()
         }
+        assertEquals("Parent job is Completed", secondException.message)
     }
 }