Do not split transactions when uploading data (#175)

* Do not split transactions when uploading data

* Add test and changelog entry

---------

Co-authored-by: Simon Binder <[email protected]>

Author: Radiokot

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## 1.0.0-BETA31
 
 * Added helpers for Attachment syncing.
+* Fix `getNextCrudTransaction()` only returning a single item.
 
 ## 1.0.0-BETA30
 

core/src/commonIntegrationTest/kotlin/com/powersync/DatabaseTest.kt

@@ -369,4 +369,25 @@ class DatabaseTest {
             val count = database.get("SELECT COUNT(*) from people") { it.getLong(0)!! }
             count shouldBe 1
         }
+
+    @Test
+    fun testCrudTransaction() =
+        databaseTest {
+            database.execute("INSERT INTO users (id, name, email) VALUES (uuid(), ?, ?)", listOf("a", "[email protected]"))
+
+            database.writeTransaction {
+                it.execute("INSERT INTO users (id, name, email) VALUES (uuid(), ?, ?)", listOf("b", "[email protected]"))
+                it.execute("INSERT INTO users (id, name, email) VALUES (uuid(), ?, ?)", listOf("c", "[email protected]"))
+            }
+
+            var transaction = database.getNextCrudTransaction()
+            transaction!!.crud shouldHaveSize 1
+            transaction.complete(null)
+
+            transaction = database.getNextCrudTransaction()
+            transaction!!.crud shouldHaveSize 2
+            transaction.complete(null)
+
+            database.getNextCrudTransaction() shouldBe null
+        }
 }

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

@@ -14,6 +14,11 @@ internal interface BucketStorage {
 
     fun nextCrudItem(transaction: PowerSyncTransaction): CrudEntry?
 
+    fun getCrudItemsByTransactionId(
+        transactionId: Int,
+        transaction: PowerSyncTransaction,
+    ): List<CrudEntry>
+
     suspend fun hasCrud(): Boolean
 
     fun hasCrud(transaction: PowerSyncTransaction): Boolean

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

@@ -41,13 +41,24 @@ internal class BucketStorageImpl(
         return id ?: throw IllegalStateException("Client ID not found")
     }
 
-    override suspend fun nextCrudItem(): CrudEntry? = db.getOptional(sql = nextCrudQuery, mapper = nextCrudMapper)
+    override suspend fun nextCrudItem(): CrudEntry? = db.getOptional(sql = nextCrudQuery, mapper = crudEntryMapper)
 
     override fun nextCrudItem(transaction: PowerSyncTransaction): CrudEntry? =
-        transaction.getOptional(sql = nextCrudQuery, mapper = nextCrudMapper)
+        transaction.getOptional(sql = nextCrudQuery, mapper = crudEntryMapper)
+
+    override fun getCrudItemsByTransactionId(
+        transactionId: Int,
+        transaction: PowerSyncTransaction,
+    ): List<CrudEntry> =
+        transaction.getAll(
+            sql = transactionCrudQuery,
+            parameters = listOf(transactionId),
+            mapper = crudEntryMapper,
+        )
 
     private val nextCrudQuery = "SELECT id, tx_id, data FROM ${InternalTable.CRUD} ORDER BY id ASC LIMIT 1"
-    private val nextCrudMapper: (SqlCursor) -> CrudEntry = { cursor ->
+    private val transactionCrudQuery = "SELECT id, tx_id, data FROM ${InternalTable.CRUD} WHERE tx_id = ? ORDER BY id ASC"
+    private val crudEntryMapper: (SqlCursor) -> CrudEntry = { cursor ->
         CrudEntry.fromRow(
             CrudRow(
                 id = cursor.getString(0)!!,

core/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt

@@ -298,15 +298,10 @@ internal class PowerSyncDatabaseImpl(
                 if (txId == null) {
                     listOf(entry)
                 } else {
-                    transaction.getAll("SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT 1") {
-                        CrudEntry.fromRow(
-                            CrudRow(
-                                id = it.getString("id"),
-                                data = it.getString("data"),
-                                txId = it.getLongOptional("tx_id")?.toInt(),
-                            ),
-                        )
-                    }
+                    bucketStorage.getCrudItemsByTransactionId(
+                        transactionId = txId,
+                        transaction = transaction,
+                    )
                 }
 
             return@readTransaction CrudTransaction(

gradle.properties

@@ -17,7 +17,7 @@ development=true
 RELEASE_SIGNING_ENABLED=true
 # Library config
 GROUP=com.powersync
-LIBRARY_VERSION=1.0.0-BETA30
+LIBRARY_VERSION=1.0.0-BETA31
 GITHUB_REPO=https://github.com/powersync-ja/powersync-kotlin.git
 # POM
 POM_URL=https://github.com/powersync-ja/powersync-kotlin/