Skip to content

Commit 7d1fa8a

Browse files
author
Oleksandr Dzhychko
committed
fix(model-server): respond with all objects if no base version was given
In an optimization, only some of the referenced objects were emitted.
1 parent 714e02d commit 7d1fa8a

File tree

4 files changed

+171
-33
lines changed

4 files changed

+171
-33
lines changed

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

Lines changed: 29 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -337,31 +337,6 @@ class ModelClientV2(
337337
httpClient.close()
338338
}
339339

340-
private suspend fun HttpResponse.readVersionDelta(): VersionDeltaStream {
341-
return if (contentType()?.match(VersionDeltaStream.CONTENT_TYPE) == true) {
342-
val content = bodyAsChannel()
343-
val versionHash = checkNotNull(content.readUTF8Line()) { "No objects received" }
344-
val versionObject = content.readUTF8Line()
345-
return if (versionObject == null) {
346-
VersionDeltaStream(versionHash, emptyFlow())
347-
} else {
348-
VersionDeltaStream(
349-
versionHash,
350-
flow {
351-
emit(versionHash to versionObject)
352-
while (true) {
353-
val key = content.readUTF8Line() ?: break
354-
val value = checkNotNull(content.readUTF8Line()) { "Object missing for hash $key" }
355-
emit(key to value)
356-
}
357-
},
358-
)
359-
}
360-
} else {
361-
body<VersionDelta>().asStream()
362-
}
363-
}
364-
365340
private suspend fun createVersion(baseVersion: CLVersion?, delta: VersionDeltaStream): CLVersion {
366341
delta.getObjectsAsFlow().collect {
367342
HashUtil.checkObjectHash(it.first, it.second)
@@ -405,10 +380,6 @@ class ModelClientV2(
405380
}
406381
}
407382

408-
private fun HttpRequestBuilder.useVersionStreamFormat() {
409-
headers.set(HttpHeaders.Accept, VersionDeltaStream.CONTENT_TYPE.toString())
410-
}
411-
412383
companion object {
413384
private val LOG = mu.KotlinLogging.logger {}
414385
fun builder(): ModelClientV2Builder = ModelClientV2PlatformSpecificBuilder()
@@ -507,6 +478,35 @@ private fun URLBuilder.appendPathSegmentsEncodingSlash(vararg components: String
507478

508479
fun VersionDelta.getAllObjects(): Map<String, String> = objectsMap + objects.associateBy { HashUtil.sha256(it) }
509480

481+
suspend fun HttpResponse.readVersionDelta(): VersionDeltaStream {
482+
return if (contentType()?.match(VersionDeltaStream.CONTENT_TYPE) == true) {
483+
val content = bodyAsChannel()
484+
val versionHash = checkNotNull(content.readUTF8Line()) { "No objects received" }
485+
val versionObject = content.readUTF8Line()
486+
return if (versionObject == null) {
487+
VersionDeltaStream(versionHash, emptyFlow())
488+
} else {
489+
VersionDeltaStream(
490+
versionHash,
491+
flow {
492+
emit(versionHash to versionObject)
493+
while (true) {
494+
val key = content.readUTF8Line() ?: break
495+
val value = checkNotNull(content.readUTF8Line()) { "Object missing for hash $key" }
496+
emit(key to value)
497+
}
498+
},
499+
)
500+
}
501+
} else {
502+
body<VersionDelta>().asStream()
503+
}
504+
}
505+
506+
fun HttpRequestBuilder.useVersionStreamFormat() {
507+
headers.set(HttpHeaders.Accept, VersionDeltaStream.CONTENT_TYPE.toString())
508+
}
509+
510510
/**
511511
* Performs a write transaction on the root node of the given branch.
512512
*

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

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ import org.modelix.model.lazy.KVEntryReference
3737
import org.modelix.model.lazy.RepositoryId
3838
import org.modelix.model.lazy.computeDelta
3939
import org.modelix.model.metameta.MetaModelBranch
40+
import org.modelix.model.persistent.CPVersion
4041
import org.modelix.model.server.store.IStoreClient
4142
import org.modelix.model.server.store.LocalModelClient
4243
import org.modelix.model.server.store.pollEntry
@@ -236,16 +237,23 @@ class RepositoriesManager(val client: LocalModelClient) {
236237
val version = CLVersion(versionHash, objectStore)
237238
// Use a bulk query to make as few request to the underlying store as possible.
238239
val bulkQuery = objectStore.newBulkQuery()
240+
// It is unsatisfactory that we have to keep already emitted hashes in memory.
241+
// But without changing the underlying model,
242+
// we have to do this to not emit objects more than once.
243+
val seenHashes = mutableSetOf<String>()
239244
fun emitObjects(entry: KVEntryReference<*>) {
240245
bulkQuery.get(entry).onSuccess {
241246
channel.trySend(entry.getHash() to it!!.serialize())
242-
for (referencedEntry in it!!.getReferencedEntries()) {
243-
emitObjects(referencedEntry)
247+
for (referencedEntry in it.getReferencedEntries()) {
248+
val wasSeenBefore = !seenHashes.add(referencedEntry.getHash())
249+
// Do not emit the object if we already emitted it.
250+
if (!wasSeenBefore) {
251+
emitObjects(referencedEntry)
252+
}
244253
}
245254
}
246255
}
247-
channel.send(versionHash to kvStore.get(versionHash)!!)
248-
emitObjects(version.treeHash!!)
256+
emitObjects(KVEntryReference(versionHash, CPVersion.DESERIALIZER))
249257
(bulkQuery as? BulkQuery)?.process()
250258
}
251259
}

‎model-server/src/test/kotlin/org/modelix/model/server/ModelClientV2Test.kt‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,10 +28,12 @@ import org.modelix.model.api.IConceptReference
2828
import org.modelix.model.api.ITree
2929
import org.modelix.model.api.PBranch
3030
import org.modelix.model.client2.ModelClientV2
31+
import org.modelix.model.client2.runWrite
3132
import org.modelix.model.lazy.CLTree
3233
import org.modelix.model.lazy.CLVersion
3334
import org.modelix.model.lazy.RepositoryId
3435
import org.modelix.model.operations.OTBranch
36+
import org.modelix.model.persistent.IKVValue
3537
import org.modelix.model.server.handlers.ModelReplicationServer
3638
import org.modelix.model.server.store.InMemoryStoreClient
3739
import org.modelix.modelql.core.count
@@ -184,4 +186,31 @@ class ModelClientV2Test {
184186
assertFalse(success)
185187
assertFalse(containsRepository)
186188
}
189+
190+
@Test
191+
fun `pulling existing versions pulls all referenced objects`() = runTest {
192+
// Arrange
193+
val url = "http://localhost/v2"
194+
val modelClientForArrange = ModelClientV2.builder().url(url).client(client).build().also { it.init() }
195+
val modelClientForAssert = ModelClientV2.builder().url(url).client(client).build().also { it.init() }
196+
val repositoryId = RepositoryId("repo1")
197+
val branchId = repositoryId.getBranchReference("my-branch")
198+
modelClientForArrange.runWrite(branchId) { root ->
199+
root.addNewChild("aChild", -1, null as IConceptReference?)
200+
}
201+
202+
// Act
203+
val versionPulled = modelClientForAssert.pullIfExists(branchId)!! as CLVersion
204+
205+
// Assert
206+
fun checkAllReferencedEntriesExistInStore(referencingEntry: IKVValue) {
207+
for (entryReference in referencingEntry.getReferencedEntries()) {
208+
// Check that the store also provides each referenced KVEntry.
209+
// `getValue` would fail if this is not the case.
210+
val referencedEntry = entryReference.getValue(versionPulled.store)
211+
checkAllReferencedEntriesExistInStore(referencedEntry)
212+
}
213+
}
214+
checkAllReferencedEntriesExistInStore(versionPulled.data!!)
215+
}
187216
}

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

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
/*
2+
* Licensed under the Apache License, Version 2.0 (the "License");
3+
* you may not use this file except in compliance with the License.
4+
* You may obtain a copy of the License at
5+
*
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
*
8+
* Unless required by applicable law or agreed to in writing,
9+
* software distributed under the License is distributed on an
10+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
11+
* KIND, either express or implied. See the License for the
12+
* specific language governing permissions and limitations
13+
* under the License.
14+
*/
15+
16+
package org.modelix.model.server.handlers
17+
18+
import io.ktor.client.request.get
19+
import io.ktor.http.appendPathSegments
20+
import io.ktor.http.takeFrom
21+
import io.ktor.serialization.kotlinx.json.json
22+
import io.ktor.server.application.install
23+
import io.ktor.server.plugins.contentnegotiation.ContentNegotiation
24+
import io.ktor.server.resources.Resources
25+
import io.ktor.server.routing.IgnoreTrailingSlash
26+
import io.ktor.server.testing.ApplicationTestBuilder
27+
import io.ktor.server.testing.testApplication
28+
import io.ktor.server.websocket.WebSockets
29+
import kotlinx.coroutines.CoroutineScope
30+
import kotlinx.coroutines.coroutineScope
31+
import org.modelix.authorization.installAuthentication
32+
import org.modelix.model.api.IConceptReference
33+
import org.modelix.model.client2.ModelClientV2
34+
import org.modelix.model.client2.readVersionDelta
35+
import org.modelix.model.client2.runWrite
36+
import org.modelix.model.client2.useVersionStreamFormat
37+
import org.modelix.model.lazy.RepositoryId
38+
import org.modelix.model.server.store.InMemoryStoreClient
39+
import org.modelix.model.server.store.LocalModelClient
40+
import kotlin.test.Test
41+
import kotlin.test.fail
42+
43+
class ModelReplicationServerTest {
44+
45+
private fun runTest(block: suspend ApplicationTestBuilder.(scope: CoroutineScope) -> Unit) = testApplication {
46+
application {
47+
installAuthentication(unitTestMode = true)
48+
install(ContentNegotiation) {
49+
json()
50+
}
51+
install(WebSockets)
52+
install(Resources)
53+
install(IgnoreTrailingSlash)
54+
val repositoriesManager = RepositoriesManager(LocalModelClient(InMemoryStoreClient()))
55+
ModelReplicationServer(repositoriesManager).init(this)
56+
KeyValueLikeModelServer(repositoriesManager).init(this)
57+
}
58+
59+
coroutineScope {
60+
block(this)
61+
}
62+
}
63+
64+
@Test
65+
fun `pulling delta does not return objects twice`() = runTest {
66+
// Arrange
67+
val url = "http://localhost/v2"
68+
val modelClient = ModelClientV2.builder().url(url).client(client).build().also { it.init() }
69+
val repositoryId = RepositoryId("repo1")
70+
val branchId = repositoryId.getBranchReference("my-branch")
71+
// By calling modelClient.runWrite twice, we create to versions.
72+
// Those two versions will share objects.
73+
modelClient.runWrite(branchId) { root ->
74+
root.addNewChild("aChild", -1, null as IConceptReference?)
75+
}
76+
modelClient.runWrite(branchId) { root ->
77+
root.addNewChild("aChild", -1, null as IConceptReference?)
78+
}
79+
80+
// Act
81+
val response = client.get {
82+
url {
83+
takeFrom(url)
84+
appendPathSegments("repositories", repositoryId.id, "branches", branchId.branchName)
85+
}
86+
useVersionStreamFormat()
87+
}
88+
val versionDelta = response.readVersionDelta()
89+
90+
// Assert
91+
val seenHashes = mutableSetOf<String>()
92+
versionDelta.getObjectsAsFlow().collect { (hash, _) ->
93+
val wasSeenBefore = !seenHashes.add(hash)
94+
if (wasSeenBefore) {
95+
// We should not send the same object (with the same hash more than once)
96+
// even if we got with versions that share data.
97+
fail("Hash $hash sent more than once.")
98+
}
99+
}
100+
}
101+
}

0 commit comments

Comments
 (0)