Adopt existing rows function

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 <R> 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<EntityId>,
+    ): Int
+
     /**
      * Asks the workers to synchronise now instead of waiting for the next trigger.
      *

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.
      *

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<SyncLimits?>(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

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<EntityId>,
+    ): 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()
     }

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<EntityId>,
+        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,

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<Unit> {
+            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<Unit> {
+            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<Unit> {
+            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<Unit> {

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
+    }
 }
 
 /**

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.