From d6ad538efd35a371f5dc7bef326eb9b9b85a26e7 Mon Sep 17 00:00:00 2001 From: Gary Bezruchko Date: Fri, 9 Oct 2026 21:46:27 +0200 Subject: [PATCH] Adopt existing rows function --- .../reflector/sync/core/CollectionHandle.kt | 43 +++++++++++ .../reflector/sync/core/log/SyncLogEvent.kt | 8 ++ .../reflector/sync/engine/CollectionWorker.kt | 27 ++++++- .../sync/engine/DefaultCollectionHandle.kt | 13 ++++ .../engine/mutation/MutationCoordinator.kt | 63 ++++++++++++++++ .../reflector/sync/engine/SyncEngineTest.kt | 75 +++++++++++++++++++ .../voir/reflector/sync/engine/TestDoubles.kt | 9 ++- docs/client-guide.md | 24 +++++- docs/sync-client-design.md | 20 +++++ 9 files changed, 277 insertions(+), 5 deletions(-) diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/CollectionHandle.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/CollectionHandle.kt index de087bd..f5c9430 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/CollectionHandle.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/CollectionHandle.kt @@ -7,6 +7,8 @@ import dev.voir.reflector.sync.core.conflict.ConflictId import dev.voir.reflector.sync.core.conflict.Resolution import dev.voir.reflector.sync.core.diagnostics.CollectionDiagnostics import dev.voir.reflector.sync.protocol.BlobId +import dev.voir.reflector.sync.protocol.EntityId +import dev.voir.reflector.sync.protocol.EntityType import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.StateFlow @@ -155,6 +157,47 @@ public interface CollectionHandle { */ public suspend fun mutate(block: suspend MutationScope.() -> R): R + /** + * Takes rows the application already holds, and the library has never been told about, into + * synchronisation, so that each is sent to the server as created. + * + * For data that existed before there was anybody to synchronise it for — a device used without + * an account whose owner now signs in, a store migrated in from elsewhere. [mutate] is the + * wrong tool there: a block is one group, and a group cannot be split without breaking the + * atomicity it was made for, so a block marking every row of a large table would be refused as + * too large and would then block the queue behind it. + * + * These rows were never changed together, so nothing ties them to each other. They are queued + * as several groups, each within the server's limit on operations per group, in the order + * given — which is the order they reach the server, so a row another one refers to belongs + * earlier in the list, or in an earlier call. All of them are queued in one transaction, so a + * process that dies part-way leaves none of them queued rather than some. + * + * Only rows the library has no record of are taken. A row it already tracks — synchronised, + * waiting to be sent, or pulled from the server — is left exactly as it is, which makes a + * repeated call with the same identifiers queue nothing the second time. + * + * The limit is the server's, so this waits until this device has heard it, which the first + * cycle after the collection is opened asks for. Offline, or with credentials the server no + * longer accepts, that wait lasts until the connection is back; a caller that cannot wait that + * long bounds it with a timeout of its own. Nothing is queued before the wait ends. + * + * Bodies are read through the adapter at push time, as for every other change, and the files + * each document names are declared with it. A row whose identifier the server already holds is + * a conflict, settled as any other: the library cannot tell one from a row it never saw. + * + * Must not be called from inside the adapter's callbacks: it may wait on a cycle, and they run + * inside that cycle. + * + * @param entityType Type of every row in [ids]. + * @param ids Rows to take in, in the order they should reach the server. Duplicates count once. + * @return How many rows were queued: those the library had no record of. + */ + public suspend fun adopt( + entityType: EntityType, + ids: List, + ): Int + /** * Asks the workers to synchronise now instead of waiting for the next trigger. * diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/log/SyncLogEvent.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/log/SyncLogEvent.kt index c5ce114..f9cbbdb 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/log/SyncLogEvent.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/core/log/SyncLogEvent.kt @@ -86,6 +86,14 @@ public enum class SyncLogEvent( */ GROUPS_MERGED(SyncLogSource.MUTATION), + /** + * Rows the application already held were taken into synchronisation. + * + * Reported at `INFO` with how many were queued and how many were already tracked: a queue that + * suddenly holds thousands of groups was put there on purpose, and this is where it says so. + */ + ENTITIES_ADOPTED(SyncLogSource.MUTATION), + /** * A file named by a document was looked into for the first time. * diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/CollectionWorker.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/CollectionWorker.kt index 81db4a7..09ce36e 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/CollectionWorker.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/CollectionWorker.kt @@ -47,6 +47,8 @@ import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.flow.filterNotNull +import kotlinx.coroutines.flow.first import kotlin.concurrent.Volatile import kotlin.coroutines.AbstractCoroutineContextElement import kotlin.coroutines.CoroutineContext @@ -143,7 +145,11 @@ internal class CollectionWorker( private val conflicts = ConflictCoordinator(scope, collection, stores, transactions, adapter, log, newUuid, references) - private var limits: SyncLimits? = null + /** + * The server's limits once heard, `null` before. Observable so that [awaitLimits] can wait for + * the cycle that asks for them rather than asking a second time on its own. + */ + private val limits = MutableStateFlow(null) private var clientId: ClientId? = null /** @@ -781,10 +787,10 @@ internal class CollectionWorker( * @return The limits, or `null` when they could not be read and the cycle has to end here. */ private suspend fun limits(): SyncLimits? { - limits?.let { return it } + limits.value?.let { return it } return try { transport.limits().also { - limits = it + limits.value = it // Server-wide rather than the scope's, so it says the server answered and nothing // about whether this user may still read the scope. recordSuccess(provesAccess = false) @@ -798,6 +804,21 @@ internal class CollectionWorker( } } + /** + * Waits until the server's limits are known, asking for a cycle if they are not yet. + * + * The cycle is what reads them, so this never sends a request of its own: two callers asking + * at once share one answer, and a failure is reported once, by the cycle, in the usual way. + * While the server cannot be reached the wait lasts; the worker keeps retrying on its triggers. + * + * @return The limits the server published. + */ + suspend fun awaitLimits(): SyncLimits { + limits.value?.let { return it } + requestSync() + return limits.filterNotNull().first() + } + private suspend fun clientId(): ClientId { clientId?.let { return it } return transactions diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/DefaultCollectionHandle.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/DefaultCollectionHandle.kt index 6828508..19bc1d9 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/DefaultCollectionHandle.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/DefaultCollectionHandle.kt @@ -23,6 +23,8 @@ import dev.voir.reflector.sync.persistence.conflict.StoredConflict import dev.voir.reflector.sync.persistence.group.PendingGroup import dev.voir.reflector.sync.protocol.BlobId import dev.voir.reflector.sync.protocol.CollectionId +import dev.voir.reflector.sync.protocol.EntityId +import dev.voir.reflector.sync.protocol.EntityType import dev.voir.reflector.sync.protocol.ScopeId import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.Flow @@ -128,6 +130,17 @@ internal class DefaultCollectionHandle( return result } + override suspend fun adopt( + entityType: EntityType, + ids: List, + ): Int { + if (ids.isEmpty()) return 0 + val limits = worker.awaitLimits() + val adopted = mutations.adopt(scope, collection, entityType, ids, limits.maxOperationsPerGroup) + if (adopted > 0) worker.requestSync() + return adopted + } + override suspend fun requestSync() { worker.requestSync() } diff --git a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/mutation/MutationCoordinator.kt b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/mutation/MutationCoordinator.kt index 0d5adb9..6ecd1bb 100644 --- a/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/mutation/MutationCoordinator.kt +++ b/client-sdk/src/commonMain/kotlin/dev/voir/reflector/sync/engine/mutation/MutationCoordinator.kt @@ -5,7 +5,10 @@ import dev.voir.reflector.sync.core.log.SyncLogger import dev.voir.reflector.sync.persistence.SyncStores import dev.voir.reflector.sync.persistence.SyncTransactionRunner import dev.voir.reflector.sync.persistence.group.PushGroupState +import dev.voir.reflector.sync.persistence.record.MutationIntent import dev.voir.reflector.sync.protocol.CollectionId +import dev.voir.reflector.sync.protocol.EntityId +import dev.voir.reflector.sync.protocol.EntityType import dev.voir.reflector.sync.protocol.GroupId import dev.voir.reflector.sync.protocol.ScopeId @@ -55,6 +58,66 @@ internal class MutationCoordinator( result } + /** + * Queues rows the library has never been told about, as groups of at most [groupSize]. + * + * One transaction for every group: either all of the rows are queued or none of them is, so a + * caller repeating the call after a crash finds nothing half done. Rows that already have a + * record are skipped rather than marked, which is what makes the repeat queue nothing — marking + * them would merge their groups into these and could make a group too large to send. + * + * @param scope Scope of the collection. + * @param collection Collection the rows belong to. + * @param entityType Type of every row. + * @param ids Rows to queue, in the order they should be sent. + * @param groupSize Most operations one group may hold: the server's limit. + * @return How many rows were queued. + */ + suspend fun adopt( + scope: ScopeId, + collection: CollectionId, + entityType: EntityType, + ids: List, + groupSize: Int, + ): Int { + require(groupSize > 0) { "a group must be able to hold at least one operation" } + val adopted = + transactions.transaction { + val generation = stores.collections.ensure(scope, collection).generation + val untracked = + ids.distinct().filter { id -> stores.records.find(scope, collection, entityType, id) == null } + for (chunk in untracked.chunked(groupSize)) { + val groupId = newGroupId() + stores.groups.create(scope, collection, groupId) + for (id in chunk) { + stores.records.markMutation( + scope = scope, + collection = collection, + entityType = entityType, + entityId = id, + intent = MutationIntent.UPSERT, + groupId = groupId, + generation = generation, + ) + } + } + untracked.size + } + // Said once per call, with the numbers: the answer to "where did this queue come from". + log.info( + SyncLogEvent.ENTITIES_ADOPTED, + context = { + mapOf( + "entityType" to entityType.value, + "adopted" to adopted.toString(), + "skipped" to (ids.size - adopted).toString(), + "groupSize" to groupSize.toString(), + ) + }, + ) { "rows the library had never been told about were queued to be sent as created" } + return adopted + } + private suspend fun queue( scope: ScopeId, collection: CollectionId, diff --git a/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/SyncEngineTest.kt b/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/SyncEngineTest.kt index 4c372b1..5adfc6a 100644 --- a/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/SyncEngineTest.kt +++ b/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/SyncEngineTest.kt @@ -33,6 +33,7 @@ import dev.voir.reflector.sync.protocol.Cursor import dev.voir.reflector.sync.protocol.EntityId import dev.voir.reflector.sync.protocol.EntityType import dev.voir.reflector.sync.protocol.EntityVersion +import dev.voir.reflector.sync.protocol.push.PushOperation import dev.voir.reflector.sync.protocol.ScopeId import dev.voir.reflector.sync.protocol.changes.ChangeBatch import dev.voir.reflector.sync.protocol.changes.ChangesPage @@ -56,6 +57,8 @@ import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.first import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.async +import kotlinx.coroutines.withTimeout import kotlinx.coroutines.withTimeoutOrNull import kotlinx.serialization.json.buildJsonObject import kotlinx.serialization.json.put @@ -588,6 +591,78 @@ class SyncEngineTest { assertTrue(sent.all { it.baseVersion == null }, "the edit claims no server state as its base") } + @Test + fun `adopted rows are sent as created, in order, in groups within the server's limit`() = + runBlocking { + transport.limits = transport.limits.copy(maxOperationsPerGroup = 2) + transport.onSnapshot = { emptySnapshot() } + transport.onPush = ::applyingEverything + val collection = engine(workers).scope(scopeId).collection(ledger) + // Rows the application wrote before there was anybody to synchronise them for. + val ids = (1..5).map { n -> EntityId(Uuid.parse("00000000-0000-7000-8000-20000000000$n")) } + ids.forEach { id -> adapter.bodies[wallet to id] = buildJsonObject { put("title", "Wallet $id") } } + + val adopted = collection.adopt(wallet, ids) + + assertEquals(5, adopted) + // Counted from the log rather than read from the state, which may still describe the + // moment before the rows were queued. + awaitRecords("three groups applied") { + logs.records.value.count { it.event == SyncLogEvent.PUSH_APPLIED } == 3 + } + val groups = transport.pushes.map { push -> push.groups.single().ops.map { it.id } } + assertEquals(listOf(ids.take(2), ids.subList(2, 4), ids.drop(4)), groups) + assertTrue( + transport.pushes.flatMap { it.groups.single().ops }.all { it is PushOperation.Upsert }, + "every row is sent as created", + ) + awaitRecord("the adoption being reported") { it.event == SyncLogEvent.ENTITIES_ADOPTED } + } + + @Test + fun `adopting leaves tracked rows alone, so repeating it queues nothing`() = + runBlocking { + transport.onSnapshot = { walletSnapshot("Cash") } + val collection = engine(workers).scope(scopeId).collection(ledger) + collection.state.await("a live collection") { it.phase == SyncPhase.LIVE } + goOffline() + val newId = EntityId(Uuid.parse("00000000-0000-7000-8000-300000000001")) + adapter.bodies[wallet to newId] = buildJsonObject { put("title", "Savings") } + + // The pulled wallet is the server's already; only the new one is taken. + assertEquals(1, collection.adopt(wallet, listOf(walletId, newId, newId))) + assertEquals(0, collection.adopt(wallet, listOf(walletId, newId))) + + val diagnostics = collection.diagnostics() + assertEquals(1, diagnostics.queue.size, "one group, not one per call") + assertEquals(1, diagnostics.pendingCount) + val pulled = assertNotNull(stores.records.find(scopeId, ledger, wallet, walletId)) + assertEquals(EntityVersion("1"), pulled.serverVersion, "the tracked row is untouched") + } + + @Test + fun `adopting waits for the server's limits and queues nothing before them`() = + runBlocking { + transport.limitsUnreachable = true + transport.onSnapshot = { emptySnapshot() } + transport.onPush = ::applyingEverything + val collection = engine(workers).scope(scopeId).collection(ledger) + adapter.bodies[wallet to walletId] = buildJsonObject { put("title", "Cash") } + + val adoption = async { collection.adopt(wallet, listOf(walletId)) } + awaitRecord("a cycle that could not read the limits") { it.event == SyncLogEvent.REQUEST_FAILED } + + assertTrue(adoption.isActive, "nothing can be grouped before the limit is known") + assertNull(stores.records.find(scopeId, ledger, wallet, walletId), "and nothing was queued") + + transport.limitsUnreachable = false + collection.requestSync() + + assertEquals(1, withTimeout(AWAIT_TIMEOUT_MILLIS) { adoption.await() }) + awaitRecord("the adopted row applied") { it.event == SyncLogEvent.PUSH_APPLIED } + assertEquals(walletId, transport.pushes.single().groups.single().ops.single().id) + } + @Test fun `a queue serving a backoff is not reported as blocked`() = runBlocking { diff --git a/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/TestDoubles.kt b/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/TestDoubles.kt index 2108bb3..e00a6e6 100644 --- a/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/TestDoubles.kt +++ b/client-sdk/src/jvmTest/kotlin/dev/voir/reflector/sync/engine/TestDoubles.kt @@ -10,6 +10,7 @@ import dev.voir.reflector.sync.core.conflict.Resolution import dev.voir.reflector.sync.core.metrics.SyncMetricEvent import dev.voir.reflector.sync.core.metrics.SyncMetrics import dev.voir.reflector.sync.core.transport.SyncTransport +import dev.voir.reflector.sync.core.transport.SyncTransportFailure import dev.voir.reflector.sync.protocol.CollectionEpoch import dev.voir.reflector.sync.protocol.CollectionId import dev.voir.reflector.sync.protocol.Cursor @@ -90,7 +91,13 @@ class FakeTransport : SyncTransport { limit: Int, ): SnapshotPage = onSnapshot(page) - override suspend fun limits(): SyncLimits = limits + /** Whether asking for the limits fails as though the server could not be reached. */ + var limitsUnreachable: Boolean = false + + override suspend fun limits(): SyncLimits { + if (limitsUnreachable) throw SyncTransportFailure.Unreachable("offline") + return limits + } } /** diff --git a/docs/client-guide.md b/docs/client-guide.md index e2b0f0c..1577a11 100644 --- a/docs/client-guide.md +++ b/docs/client-guide.md @@ -305,6 +305,27 @@ entity outlives your row: it carries the acknowledged version, which is what sup your own deletion when it comes back through the log. It is cleared by the next bootstrap's sweep, where the entity is simply absent from the snapshot. +### Rows you had before there was anybody to sync them for + +A device used without an account, whose owner then signs in, holds rows the library was never told +about — and never will be by `mutate`, which you would have to call once per row or once for the +lot. Once per row is thousands of round trips. Once for the lot is one group, refused as too large +before it leaves the device, and blocking everything queued behind it. + +```kotlin +val scope = engine.scope(ScopeId(ownerId)) +val ledger = scope.collection(LEDGER) + +ledger.adopt(LedgerAdapter.WALLET, walletIds) // referred to, so first +ledger.adopt(LedgerAdapter.TRANSACTION, transactionIds) // queued behind the wallets +``` + +Each call queues the rows as groups within the server's limit, in the order you gave, and sends +them as created. It takes only rows the library has no record of, so calling it again after a crash +queues nothing twice. It waits until the server's limit is known — the first cycle after opening the +collection reads it — so call it after `collection(...)`, online, and bound it with a timeout if a +screen is waiting on it. + ## Reading state ```kotlin @@ -543,7 +564,8 @@ adapter. The whole of it — bindings, fetch policies and what your screens have 2. KSP configured per target. 3. An adapter per collection, with no transactions and no I/O inside it. 4. `syncHttpClient` (or a transport that keeps `explicitNulls` and non-throwing status codes). -5. Every write inside `mutate { }`, with an explicit `markUpserted` / `markDeleted`. +5. Every write inside `mutate { }`, with an explicit `markUpserted` / `markDeleted` — and rows that + existed before the scope did taken in with `adopt`. 6. A `TokenProvider` that can actually refresh. 7. Trigger sources for foreground and connectivity — the timer alone is a floor, not a plan. 8. A resync path for your own schema migrations. diff --git a/docs/sync-client-design.md b/docs/sync-client-design.md index 9260ac6..0c621f5 100644 --- a/docs/sync-client-design.md +++ b/docs/sync-client-design.md @@ -188,6 +188,25 @@ reacting looks like from here: correcting a refused entity moves it into a later leaves behind is a shell whose records are gone. Without that step, refusing to look past the head would block the collection permanently. +**Rows that existed before synchronisation: `adopt`.** Data the application held before there was +an owner to synchronise it for — a device used without an account whose owner now signs in — has +no metadata row, so nothing would ever send it. Marking all of it inside one `mutate` would be +correct and useless: one block is one group, a group is never split, and a group holding a whole +table is refused locally as too large and then blocks the queue behind it. + +`adopt(entityType, ids)` is the other shape. Those rows were never changed together, so nothing +ties them, and they are queued as consecutive groups of at most `maxOperationsPerGroup`, in the +order given — the order matters for the same reason FIFO does above, so a caller passes the rows +others refer to first. Every group is created in one transaction: all or nothing, so a process +that dies part-way leaves nothing to reconcile. Only rows with no metadata row are taken; a row the +library already tracks keeps its state, which makes the call idempotent and is also why it can +never pull an existing group into these and make it too large. + +The limit is the server's, read by the first cycle. `adopt` waits for that answer rather than +guessing a size or asking on its own: guessing too high blocks the queue, guessing too low costs a +round trip per group, and a second request would race the cycle's. Until the limit is known nothing +is queued. + **The "edited while in flight" race** is closed by revisions: when the envelope is assembled `pushing_rev := local_rev`; on ack `acked_rev := pushing_rev`. If `local_rev` moved ahead while the envelope was in flight, the record stays dirty and after the ack joins a new group. @@ -818,6 +837,7 @@ interface CollectionHandle { // NEEDS_ATTENTION | RESYNC_REQUIRED val conflicts: Flow> // only what the adapter did not resolve itself suspend fun mutate(block: suspend MutationScope.() -> R): R // = withTransaction + suspend fun adopt(entityType: EntityType, ids: List): Int // pre-existing rows; §4 suspend fun requestSync() suspend fun requestResync() // drop the cursor and fetch the snapshot again val refusals: Flow> // groups refused for good, read from storage