Merge remote-tracking branch 'origin/main' into rust-sync-client

Author: simolus3

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## 1.1.2 (pending)
+
+* [Android, JVM] Use version `0.4.0` of `powersync-sqlite-core`.
+
+## 1.1.1
+
+* Fix reported progress around compactions / defrags on the sync service.
+* [Android] Set `temp_store_directory`, avoiding crashes for large materialized views.
+
 ## 1.1.0
 
 * Add `trackPreviousValues` option on `Table` which sets `CrudEntry.previousValues` to previous values on updates.

core-tests-android/src/androidTest/java/com/powersync/AndroidDatabaseTest.kt

@@ -222,4 +222,14 @@ class AndroidDatabaseTest {
             // The exception messages differ slightly between drivers
             assertEquals(exception.message!!.contains("write a readonly database"), true)
         }
+
+    @Test
+    fun canUseTempStore() = runTest {
+        database.execute("PRAGMA temp_store = 1;") // Store temporary data as files
+        database.execute("CREATE TEMP TABLE foo (bar TEXT);")
+        val data = "new row".repeat(100);
+        for (i in 0..10000) {
+            database.execute("INSERT INTO foo VALUES (?)", parameters = listOf(data))
+        }
+    }
 }

core/src/androidMain/kotlin/com/powersync/DatabaseDriverFactory.android.kt

@@ -7,6 +7,7 @@ import com.powersync.db.internal.InternalSchema
 import com.powersync.db.migrateDriver
 import kotlinx.coroutines.CoroutineScope
 import org.sqlite.SQLiteCommitListener
+import java.util.concurrent.atomic.AtomicBoolean
 
 @Suppress("EXPECT_ACTUAL_CLASSIFIERS_ARE_IN_BETA_WARNING")
 public actual class DatabaseDriverFactory(
@@ -27,10 +28,22 @@ public actual class DatabaseDriverFactory(
                 context.getDatabasePath(dbFilename)
             }
 
+        val properties = buildDefaultWalProperties(readOnly = readOnly)
+        val isFirst = IS_FIRST_CONNECTION.getAndSet(false)
+        if (isFirst) {
+            // Make sure the temp_store_directory points towards a temporary directory we actually
+            // have access to. Due to sandboxing, the default /tmp/ is inaccessible.
+            // The temp_store_directory pragma is deprecated and not thread-safe, so we only set it
+            // on the first connection (it sets a global field and will affect every connection
+            // opened).
+            val escapedPath = context.cacheDir.absolutePath.replace("\"", "\"\"")
+            properties.setProperty("temp_store_directory", "\"$escapedPath\"")
+        }
+
         val driver =
             JdbcSqliteDriver(
                 url = "jdbc:sqlite:$dbPath",
-                properties = buildDefaultWalProperties(readOnly = readOnly),
+                properties = properties,
             )
 
         migrateDriver(driver, schema)
@@ -59,4 +72,8 @@ public actual class DatabaseDriverFactory(
 
         return mappedDriver
     }
+
+    private companion object {
+        val IS_FIRST_CONNECTION = AtomicBoolean(true)
+    }
 }

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

@@ -10,10 +10,11 @@ import com.powersync.bucket.OplogEntry
 import com.powersync.testutils.ActiveDatabaseTest
 import com.powersync.testutils.databaseTest
 import com.powersync.testutils.waitFor
+import io.kotest.assertions.withClue
+import io.kotest.matchers.properties.shouldHaveValue
 import kotlinx.coroutines.channels.Channel
 import kotlin.test.BeforeTest
 import kotlin.test.Test
-import kotlin.test.assertEquals
 import kotlin.test.assertFalse
 import kotlin.test.assertNull
 import kotlin.test.assertTrue
@@ -80,6 +81,16 @@ abstract class BaseSyncProgressTest(
         }
     }
 
+    private fun ProgressWithOperations.shouldBe(
+        downloaded: Int,
+        total: Int,
+    ) {
+        withClue("progress $downloadedOperations/$totalOperations should be $downloaded/$total") {
+            this::downloadedOperations shouldHaveValue downloaded
+            this::totalOperations shouldHaveValue total
+        }
+    }
+
     private suspend fun ReceiveTurbine<SyncStatusData>.expectProgress(
         total: Pair<Int, Int>,
         priorities: Map<BucketPriority, Pair<Int, Int>> = emptyMap(),
@@ -88,14 +99,12 @@ abstract class BaseSyncProgressTest(
         val progress = item.downloadProgress ?: error("Expected download progress on $item")
 
         assertTrue { item.downloading }
-        assertEquals(total.first, progress.downloadedOperations)
-        assertEquals(total.second, progress.totalOperations)
+        progress.shouldBe(total.first, total.second)
 
         priorities.forEach { (priority, expected) ->
             val (expectedDownloaded, expectedTotal) = expected
-            val progress = progress.untilPriority(priority)
-            assertEquals(expectedDownloaded, progress.downloadedOperations)
-            assertEquals(expectedTotal, progress.totalOperations)
+            val actualProgress = progress.untilPriority(priority)
+            actualProgress.shouldBe(expectedDownloaded, expectedTotal)
         }
     }
 
@@ -278,6 +287,61 @@ abstract class BaseSyncProgressTest(
             syncLines.close()
         }
 
+    @Test
+    fun interruptedWithDefrag() =
+        databaseTest {
+            database.connect(connector)
+
+            turbineScope {
+                val turbine = database.currentStatus.asFlow().testIn(this)
+                turbine.waitFor { it.connected && !it.downloading }
+                syncLines.send(
+                    SyncLine.FullCheckpoint(
+                        Checkpoint(
+                            lastOpId = "10",
+                            checksums = listOf(bucket("a", 10)),
+                        ),
+                    ),
+                )
+                turbine.expectProgress(0 to 10)
+
+                addDataLine("a", 5)
+                turbine.expectProgress(5 to 10)
+
+                turbine.cancel()
+            }
+
+            // Close and re-connect
+            database.close()
+            syncLines.close()
+            database = openDatabase()
+            syncLines = Channel()
+            database.connect(connector)
+
+            turbineScope {
+                val turbine = database.currentStatus.asFlow().testIn(this)
+                turbine.waitFor { it.connected && !it.downloading }
+
+                // A sync rule deploy could reset buckets, making the new bucket smaller than the
+                // existing one.
+                syncLines.send(
+                    SyncLine.FullCheckpoint(
+                        Checkpoint(
+                            lastOpId = "14",
+                            checksums = listOf(bucket("a", 4)),
+                        ),
+                    ),
+                )
+
+                // In this special case, don't report 5/4 as progress
+                turbine.expectProgress(0 to 4)
+                turbine.cancel()
+            }
+
+            database.close()
+            syncLines.close()
+        }
+
     @Test
     fun differentPriorities() =
         databaseTest {

core/src/commonMain/kotlin/com/powersync/sync/Progress.kt

@@ -3,6 +3,7 @@ package com.powersync.sync
 import com.powersync.bucket.BucketPriority
 import com.powersync.bucket.Checkpoint
 import com.powersync.bucket.LocalOperationCounters
+import kotlin.math.min
 
 /**
  * Information about a progressing download.
@@ -77,11 +78,19 @@ public data class SyncDownloadProgress internal constructor(
         downloadedOperations = completed
     }
 
+    /**
+     * Creates download progress information from the local progress counters since the last full sync and the target
+     * checkpoint.
+     */
     @LegacySyncImplementation
     internal constructor(localProgress: Map<String, LocalOperationCounters>, target: Checkpoint) : this(
         buildMap {
+            var invalidated = false
+
             for (entry in target.checksums) {
                 val savedProgress = localProgress[entry.bucket]
+                val atLast = savedProgress?.atLast ?: 0
+                val sinceLast = savedProgress?.sinceLast ?: 0
 
                 put(
                     entry.bucket,
@@ -92,6 +101,22 @@ public data class SyncDownloadProgress internal constructor(
                         targetCount = (entry.count ?: 0).toLong(),
                     ),
                 )
+
+                entry.count?.let { knownCount ->
+                    if (knownCount < atLast + sinceLast) {
+                        // Either due to a defrag / sync rule deploy or a compaction operation, the
+                        // size of the bucket shrank so much that the local ops exceed the ops in
+                        // the updated bucket. We can't possibly report progress in this case (it
+                        // would overshoot 100%).
+                        invalidated = true
+                    }
+                }
+            }
+
+            if (invalidated) {
+                for ((key, value) in entries) {
+                    put(key, value.copy(sinceLast = 0, atLast = 0))
+                }
             }
         },
     )
@@ -118,7 +143,7 @@ public data class SyncDownloadProgress internal constructor(
                     put(
                         bucket.bucket,
                         previous.copy(
-                            sinceLast = previous.sinceLast + bucket.data.size,
+                            sinceLast = min(previous.sinceLast + bucket.data.size, previous.total),
                         ),
                     )
                 }

core/src/commonMain/kotlin/com/powersync/sync/SyncStream.kt

@@ -48,6 +48,7 @@ import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
 import kotlinx.datetime.Clock
+import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.JsonElement
 import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.encodeToJsonElement
@@ -635,8 +636,8 @@ internal class SyncStream(
             state: SyncStreamState,
         ): SyncStreamState {
             val batch = SyncDataBatch(listOf(data))
-            status.update { copy(downloading = true, downloadProgress = downloadProgress?.incrementDownloaded(batch)) }
             bucketStorage.saveSyncData(batch)
+            status.update { copy(downloading = true, downloadProgress = downloadProgress?.incrementDownloaded(batch)) }
             return state
         }
 

demos/hello-powersync/composeApp/build.gradle.kts

@@ -29,7 +29,7 @@ kotlin {
 
         pod("powersync-sqlite-core") {
             linkOnly = true
-            version = "0.3.12"
+            version = "0.4.0"
         }
 
         framework {

demos/supabase-todolist/gradle.properties

@@ -1,6 +1,7 @@
 kotlin.code.style=official
 xcodeproj=./iosApp
 android.useAndroidX=true
+org.gradle.caching=true
 org.gradle.jvmargs=-Xmx3g
 org.jetbrains.compose.experimental.jscanvas.enabled=true
 org.jetbrains.compose.experimental.macos.enabled=true

demos/supabase-todolist/shared/build.gradle.kts

@@ -27,7 +27,7 @@ kotlin {
         ios.deploymentTarget = "14.1"
         podfile = project.file("../iosApp/Podfile")
         pod("powersync-sqlite-core") {
-            version = "0.3.12"
+            version = "0.4.0"
             linkOnly = true
         }
 

gradle.properties

@@ -1,6 +1,7 @@
 kotlin.code.style=official
 # Gradle
 org.gradle.jvmargs=-Xmx2048M -Dfile.encoding=UTF-8 -Dkotlin.daemon.jvm.options\="-Xmx2048M"
+org.gradle.caching=true
 # Compose
 org.jetbrains.compose.experimental.uikit.enabled=true
 # Android
@@ -17,7 +18,7 @@ development=true
 RELEASE_SIGNING_ENABLED=true
 # Library config
 GROUP=com.powersync
-LIBRARY_VERSION=1.1.0
+LIBRARY_VERSION=1.1.1
 GITHUB_REPO=https://github.com/powersync-ja/powersync-kotlin.git
 # POM
 POM_URL=https://github.com/powersync-ja/powersync-kotlin/

gradle/libs.versions.toml

@@ -18,7 +18,7 @@ kotlinx-io = "0.5.4"
 ktor = "3.1.0"
 rsocket = "0.20.0"
 uuid = "0.8.2"
-powersync-core = "0.3.14"
+powersync-core = "0.4.0"
 sqlite-jdbc = "3.49.1.0"
 sqliter = "1.3.1"
 turbine = "1.2.0"