Fix sync integration tests

Author: simolus3

core/src/commonIntegrationTest/kotlin/com/powersync/sync/SyncIntegrationTest.kt

@@ -440,8 +440,15 @@ class SyncIntegrationTest {
 
             // Trigger an upload (adding a keep-alive sync line because the execute could start before the database is fully
             // connected).
+            turbineScope {
+                val turbine = database.currentStatus.asFlow().testIn(this)
+                syncLines.send(SyncLine.KeepAlive(1234))
+                turbine.waitFor { it.connected }
+                turbine.cancelAndIgnoreRemainingEvents()
+            }
+
             database.execute("INSERT INTO users (id, name, email) VALUES (uuid(), ?, ?)", listOf("local", "local@example.org"))
-            syncLines.send(SyncLine.KeepAlive(1234))
+
             expectUserRows(1)
             uploadStarted.await()
 

core/src/commonIntegrationTest/kotlin/com/powersync/testutils/TestUtils.kt

@@ -77,7 +77,7 @@ internal class ActiveDatabaseTest(
     val logger =
         Logger(
             TestConfig(
-                minSeverity = Severity.Debug,
+                minSeverity = Severity.Verbose,
                 logWriterList = listOf(logWriter, generatePrintLogWriter()),
             ),
         )

core/src/commonMain/kotlin/com/powersync/bucket/BucketStorage.kt

@@ -2,6 +2,7 @@ package com.powersync.bucket
 
 import com.powersync.db.crud.CrudEntry
 import com.powersync.db.internal.PowerSyncTransaction
+import com.powersync.sync.Instruction
 import com.powersync.sync.SyncDataBatch
 import com.powersync.sync.SyncLocalDatabaseResult
 
@@ -25,20 +26,8 @@ internal interface BucketStorage {
 
     suspend fun updateLocalTarget(checkpointCallback: suspend () -> String): Boolean
 
-    suspend fun saveSyncData(syncDataBatch: SyncDataBatch)
-
-    suspend fun getBucketStates(): List<BucketState>
-
-    suspend fun getBucketOperationProgress(): Map<String, LocalOperationCounters>
-
-    suspend fun removeBuckets(bucketsToDelete: List<String>)
-
     suspend fun hasCompletedSync(): Boolean
 
-    suspend fun syncLocalDatabase(
-        targetCheckpoint: Checkpoint,
-        partialPriority: BucketPriority? = null,
-    ): SyncLocalDatabaseResult
-
-    fun setTargetCheckpoint(checkpoint: Checkpoint)
+    suspend fun control(op: String, payload: String?): List<Instruction>
+    suspend fun control(op: String, payload: ByteArray): List<Instruction>
 }

core/src/commonMain/kotlin/com/powersync/bucket/BucketStorageImpl.kt

@@ -8,6 +8,7 @@ import com.powersync.db.crud.CrudRow
 import com.powersync.db.internal.InternalDatabase
 import com.powersync.db.internal.InternalTable
 import com.powersync.db.internal.PowerSyncTransaction
+import com.powersync.sync.Instruction
 import com.powersync.sync.SyncDataBatch
 import com.powersync.sync.SyncLocalDatabaseResult
 import com.powersync.utils.JsonUtil
@@ -21,14 +22,8 @@ internal class BucketStorageImpl(
     private var hasCompletedSync = AtomicBoolean(false)
     private var pendingBucketDeletes = AtomicBoolean(false)
 
-    /**
-     * Count up, and do a compact on startup.
-     */
-    private var compactCounter = COMPACT_OPERATION_INTERVAL
-
     companion object {
         const val MAX_OP_ID = "9223372036854775807"
-        const val COMPACT_OPERATION_INTERVAL = 1_000
     }
 
     override fun getMaxOpId(): String = MAX_OP_ID
@@ -130,50 +125,6 @@ internal class BucketStorageImpl(
         }
     }
 
-    override suspend fun saveSyncData(syncDataBatch: SyncDataBatch) {
-        db.writeTransaction { tx ->
-            val jsonString = JsonUtil.json.encodeToString(syncDataBatch)
-            tx.execute(
-                "INSERT INTO powersync_operations(op, data) VALUES(?, ?)",
-                listOf("save", jsonString),
-            )
-        }
-        this.compactCounter += syncDataBatch.buckets.sumOf { it.data.size }
-    }
-
-    override suspend fun getBucketStates(): List<BucketState> =
-        db.getAll(
-            "SELECT name AS bucket, CAST(last_op AS TEXT) AS op_id FROM ${InternalTable.BUCKETS} WHERE pending_delete = 0 AND name != '\$local'",
-            mapper = { cursor ->
-                BucketState(
-                    bucket = cursor.getString(0)!!,
-                    opId = cursor.getString(1)!!,
-                )
-            },
-        )
-
-    override suspend fun getBucketOperationProgress(): Map<String, LocalOperationCounters> =
-        buildMap {
-            val rows =
-                db.getAll("SELECT name, count_at_last, count_since_last FROM ps_buckets") { cursor ->
-                    cursor.getString(0)!! to
-                        LocalOperationCounters(
-                            atLast = cursor.getLong(1)!!.toInt(),
-                            sinceLast = cursor.getLong(2)!!.toInt(),
-                        )
-                }
-
-            for ((name, counters) in rows) {
-                put(name, counters)
-            }
-        }
-
-    override suspend fun removeBuckets(bucketsToDelete: List<String>) {
-        bucketsToDelete.forEach { bucketName ->
-            deleteBucket(bucketName)
-        }
-    }
-
     private suspend fun deleteBucket(bucketName: String) {
         db.writeTransaction { tx ->
             tx.execute(
@@ -208,202 +159,26 @@ internal class BucketStorageImpl(
         }
     }
 
-    override suspend fun syncLocalDatabase(
-        targetCheckpoint: Checkpoint,
-        partialPriority: BucketPriority?,
-    ): SyncLocalDatabaseResult {
-        val result = validateChecksums(targetCheckpoint, partialPriority)
-
-        if (!result.checkpointValid) {
-            logger.w { "[SyncLocalDatabase] Checksums failed for ${result.checkpointFailures}" }
-            result.checkpointFailures?.forEach { bucketName ->
-                deleteBucket(bucketName)
-            }
-            result.ready = false
-            return result
-        }
-
-        val bucketNames =
-            targetCheckpoint.checksums
-                .let {
-                    if (partialPriority == null) {
-                        it
-                    } else {
-                        it.filter { cs -> cs.priority >= partialPriority }
-                    }
-                }.map { it.bucket }
-
-        db.writeTransaction { tx ->
-            tx.execute(
-                "UPDATE ps_buckets SET last_op = ? WHERE name IN (SELECT json_each.value FROM json_each(?))",
-                listOf(targetCheckpoint.lastOpId, JsonUtil.json.encodeToString(bucketNames)),
-            )
-
-            if (partialPriority == null && targetCheckpoint.writeCheckpoint != null) {
-                tx.execute(
-                    "UPDATE ps_buckets SET last_op = ? WHERE name = '\$local'",
-                    listOf(targetCheckpoint.writeCheckpoint),
-                )
-            }
-        }
-
-        val valid = updateObjectsFromBuckets(targetCheckpoint, partialPriority)
-
-        if (!valid) {
-            return SyncLocalDatabaseResult(
-                ready = false,
-                checkpointValid = true,
-            )
-        }
-
-        this.forceCompact()
-
-        return SyncLocalDatabaseResult(
-            ready = true,
-        )
-    }
-
-    private suspend fun validateChecksums(
-        checkpoint: Checkpoint,
-        priority: BucketPriority? = null,
-    ): SyncLocalDatabaseResult {
-        val serializedCheckpoint =
-            JsonUtil.json.encodeToString(
-                when (priority) {
-                    null -> checkpoint
-                    // Only validate buckets with a priority included in this partial sync.
-                    else -> checkpoint.copy(checksums = checkpoint.checksums.filter { it.priority >= priority })
-                },
-            )
-
-        val res =
-            db.getOptional(
-                "SELECT powersync_validate_checkpoint(?) AS result",
-                parameters = listOf(serializedCheckpoint),
-                mapper = { cursor ->
-                    cursor.getString(0)!!
-                },
-            )
-                ?: // no result
-                return SyncLocalDatabaseResult(
-                    ready = false,
-                    checkpointValid = false,
-                )
+    private fun handleControlResult(cursor: SqlCursor): List<Instruction> {
+        val result = cursor.getString(0)!!
+        logger.v { "control result: $result" }
 
-        return JsonUtil.json.decodeFromString<SyncLocalDatabaseResult>(res)
+        return JsonUtil.json.decodeFromString<List<Instruction>>(result)
     }
 
-    /**
-     * Atomically update the local state.
-     *
-     * This includes creating new tables, dropping old tables, and copying data over from the oplog.
-     */
-    private suspend fun updateObjectsFromBuckets(
-        checkpoint: Checkpoint,
-        priority: BucketPriority? = null,
-    ): Boolean {
-        @Serializable
-        data class SyncLocalArgs(
-            val priority: BucketPriority,
-            val buckets: List<String>,
-        )
-
-        val args =
-            if (priority != null) {
-                JsonUtil.json.encodeToString(
-                    SyncLocalArgs(
-                        priority = priority,
-                        buckets = checkpoint.checksums.filter { it.priority >= priority }.map { it.bucket },
-                    ),
-                )
-            } else {
-                ""
-            }
-
+    override suspend fun control(op: String, payload: String?): List<Instruction> {
         return db.writeTransaction { tx ->
-            tx.execute(
-                "INSERT INTO powersync_operations(op, data) VALUES(?, ?)",
-                listOf("sync_local", args),
-            )
-
-            val res =
-                tx.get("select last_insert_rowid()") { cursor ->
-                    cursor.getLong(0)!!
-                }
-
-            val didApply = res == 1L
-            if (didApply && priority == null) {
-                // Reset progress counters. We only do this for a complete sync, as we want a download progress to
-                // always cover a complete checkpoint instead of resetting for partial completions.
-                tx.execute(
-                    """
-                    UPDATE ps_buckets SET count_since_last = 0, count_at_last = ?1->name
-                      WHERE ?1->name IS NOT NULL
-                    """.trimIndent(),
-                    listOf(
-                        JsonUtil.json.encodeToString(
-                            buildMap<String, Int> {
-                                for (bucket in checkpoint.checksums) {
-                                    bucket.count?.let { put(bucket.bucket, it) }
-                                }
-                            },
-                        ),
-                    ),
-                )
-            }
-
-            return@writeTransaction didApply
-        }
-    }
+            logger.v { "powersync_control($op, $payload)" }
 
-    private suspend fun forceCompact() {
-        // Reset counter
-        this.compactCounter = COMPACT_OPERATION_INTERVAL
-        this.pendingBucketDeletes.value = true
-
-        this.autoCompact()
-    }
-
-    private suspend fun autoCompact() {
-        // 1. Delete buckets
-        deletePendingBuckets()
-
-        // 2. Clear REMOVE operations, only keeping PUT ones
-        clearRemoveOps()
-    }
-
-    private suspend fun deletePendingBuckets() {
-        if (!this.pendingBucketDeletes.value) {
-            return
-        }
-
-        db.writeTransaction { tx ->
-            tx.execute(
-                "INSERT INTO powersync_operations(op, data) VALUES (?, ?)",
-                listOf("delete_pending_buckets", ""),
-            )
-
-            // Executed once after start-up, and again when there are pending deletes.
-            pendingBucketDeletes.value = false
+            tx.get("SELECT powersync_control(?, ?) AS r", listOf(op, payload), ::handleControlResult)
         }
     }
 
-    private suspend fun clearRemoveOps() {
-        if (this.compactCounter < COMPACT_OPERATION_INTERVAL) {
-            return
-        }
+    override suspend fun control(op: String, payload: ByteArray): List<Instruction> {
+        return db.writeTransaction { tx ->
+            logger.v { "powersync_control($op, binary payload)" }
 
-        db.writeTransaction { tx ->
-            tx.execute(
-                "INSERT INTO powersync_operations(op, data) VALUES (?, ?)",
-                listOf("clear_remove_ops", ""),
-            )
+            tx.get("SELECT powersync_control(?, ?) AS r", listOf(op, payload), ::handleControlResult)
         }
-        this.compactCounter = 0
-    }
-
-    @Suppress("UNUSED_PARAMETER")
-    override fun setTargetCheckpoint(checkpoint: Checkpoint) {
-        // No-op for now
     }
 }