fix: minor memory tweaks (#337)

stevensJourney · web-flow · commit 9590c1361455 · 2026-03-26T11:08:38.000+02:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -3,29 +3,37 @@
 ## 1.11.2 (unreleased)
 
 - Don't attempt to create WebSocket connections on watchOS.
+- Update default SQLite cache size to 50MB, this was previously erroneously set to 200MB
+- Move dispatching responsibility into `SQLiteConnectionPool` implementations after obtaining a
+  connection lease.
+  Implementers of `SQLiteConnectionPool` should dispatch blocking SQLite callbacks to an
+  appropriate dispatcher such as `Dispatchers.IO`. This should prevent the worker pool from
+  expanding when large numbers of concurrent operations are requested.
 
 ## 1.11.1
 
 - Fix RSocket connection bugs on iOS (and other platforms using the RSocket sync transport):
-  - Fix false `connected: true` status when using an invalid token. `ConnectionEstablished` is now
-    only emitted when the first data frame arrives from the server, matching the HTTP path which
-    waits for a `200 OK`.
-  - Fix sync loop terminating permanently when the server rejects the connection with an RSocket
-    ERROR frame (e.g. invalid JWT). `RSocketError` extends `Throwable` not `Exception`, so it was
-    not caught by the retry loop.
-  - Fix sync loop stalling indefinitely after a transport-layer failure (dead socket, network
-    dropout).
+    - Fix false `connected: true` status when using an invalid token. `ConnectionEstablished` is now
+      only emitted when the first data frame arrives from the server, matching the HTTP path which
+      waits for a `200 OK`.
+    - Fix sync loop terminating permanently when the server rejects the connection with an RSocket
+      ERROR frame (e.g. invalid JWT). `RSocketError` extends `Throwable` not `Exception`, so it was
+      not caught by the retry loop.
+    - Fix sync loop stalling indefinitely after a transport-layer failure (dead socket, network
+      dropout).
 
 ## 1.11.0
 
 - __Breaking__: On tables, the `localOnly`, `insertOnly`, `trackMetadata`, `trackPreviousValues` and
   `ignoreEmptyUpdates` options are now stored in a `TableOptions` class. Existing constructors
   continue to work, but calling `copy` with any of these parameters no longer works.
 - Make raw tables easier to use:
-  - Introduce the `RawTableSchema` class storing the name of a raw table in the database. When set,
-    `put` and `delete` statements can be inferred automatically.
-  - Add `RawTable.jsonDescription`, which can be passed to the `powersync_create_raw_table_crud_trigger`
-    SQL function to auto-create triggers forwarding writes to `ps_crud`.
+    - Introduce the `RawTableSchema` class storing the name of a raw table in the database. When
+      set,
+      `put` and `delete` statements can be inferred automatically.
+    - Add `RawTable.jsonDescription`, which can be passed to the
+      `powersync_create_raw_table_crud_trigger`
+      SQL function to auto-create triggers forwarding writes to `ps_crud`.
 - Update PowerSync core extension to version 0.4.11.
 - Remove the experimental label from Sync Stream APIs.
 - Compose: add `composeSyncStream` helper method to subscribe to Sync Streams in a composition.
@@ -59,7 +67,8 @@
 
 ## 1.10.2
 
-- Exceptions that occur while initializing a PowerSync database are now rethrown when the database is used.
+- Exceptions that occur while initializing a PowerSync database are now rethrown when the database
+  is used.
 - [Internal] Updated PowerSyncKotlin build to use SKIEE's `produceDistributableFramework`.
 
 ## 1.10.1
diff --git a/common/src/commonMain/kotlin/com/powersync/db/driver/InternalConnectionPool.kt b/common/src/commonMain/kotlin/com/powersync/db/driver/InternalConnectionPool.kt
@@ -6,11 +6,14 @@ import com.powersync.ExperimentalPowerSyncAPI
 import com.powersync.PersistentConnectionFactory
 import com.powersync.utils.JsonUtil
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.IO
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
 
 @OptIn(ExperimentalPowerSyncAPI::class)
 internal class InternalConnectionPool(
@@ -26,6 +29,11 @@ internal class InternalConnectionPool(
     // MutableSharedFlow to emit batched table updates
     private val tableUpdatesFlow = MutableSharedFlow<Set<String>>(replay = 0)
 
+    // Database calls are synchronous/blocking, so we always run them on Dispatchers.IO instead of
+    // inheriting the caller-provided scope context. The provided scope is still used for
+    // lifecycle-bound pool coroutines like read workers and update emission.
+    private val dispatcher = Dispatchers.IO
+
     private fun newConnection(readOnly: Boolean): SQLiteConnection {
         val connection =
             factory.openConnection(
@@ -38,19 +46,26 @@ internal class InternalConnectionPool(
         return connection
     }
 
-    override suspend fun <T> read(callback: suspend (SQLiteConnectionLease) -> T): T = readPool.read(callback)
+    override suspend fun <T> read(callback: suspend (SQLiteConnectionLease) -> T): T =
+        readPool.read { connection ->
+            withContext(dispatcher) {
+                callback(connection)
+            }
+        }
 
     override suspend fun <T> write(callback: suspend (SQLiteConnectionLease) -> T): T =
         writeLockMutex.withLock {
-            try {
-                callback(RawConnectionLease(writeConnection))
-            } finally {
-                // When we've leased a write connection, we may have to update table update flows
-                // after users ran their custom statements.
-                val updatedTables = writeConnection.readPendingUpdates()
-                if (updatedTables.isNotEmpty()) {
-                    scope.launch {
-                        tableUpdatesFlow.emit(updatedTables)
+            withContext(dispatcher) {
+                try {
+                    callback(RawConnectionLease(writeConnection))
+                } finally {
+                    // When we've leased a write connection, we may have to update table update flows
+                    // after users ran their custom statements.
+                    val updatedTables = writeConnection.readPendingUpdates()
+                    if (updatedTables.isNotEmpty()) {
+                        scope.launch {
+                            tableUpdatesFlow.emit(updatedTables)
+                        }
                     }
                 }
             }
@@ -61,6 +76,7 @@ internal class InternalConnectionPool(
         readPool.withAllConnections { rawReadConnections ->
             val readers = rawReadConnections.map { RawConnectionLease(it) }
             // Then get access to the write connection
+            // The write call will dispatch to the provided dispatcher
             write { writer ->
                 action(writer, readers)
             }
@@ -85,7 +101,7 @@ internal fun SQLiteConnection.setupDefaultPragmas(readOnly: Boolean) {
 
     execSQL("pragma journal_size_limit = ${6 * 1024 * 1024}")
     execSQL("pragma busy_timeout = 30000")
-    execSQL("pragma cache_size = ${50 * 1024}")
+    execSQL("pragma cache_size = -${50 * 1024}")
 
     // Older versions of the SDK used to set up an empty schema and raise the user version to 1.
     // Keep doing that for consistency.
diff --git a/common/src/commonMain/kotlin/com/powersync/db/driver/SingleConnectionPool.kt b/common/src/commonMain/kotlin/com/powersync/db/driver/SingleConnectionPool.kt
@@ -2,10 +2,13 @@ package com.powersync.db.driver
 
 import androidx.sqlite.SQLiteConnection
 import com.powersync.ExperimentalPowerSyncAPI
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.IO
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
 
 /**
  * A [SQLiteConnectionPool] backed by a single database connection.
@@ -20,6 +23,8 @@ public class SingleConnectionPool(
     private var closed = false
     private val tableUpdatesFlow = MutableSharedFlow<Set<String>>(replay = 0)
 
+    private val dispatcher = Dispatchers.IO
+
     init {
         conn.setupDefaultPragmas(false)
     }
@@ -28,14 +33,16 @@ public class SingleConnectionPool(
 
     override suspend fun <T> write(callback: suspend (SQLiteConnectionLease) -> T): T =
         mutex.withLock {
-            check(!closed) { "Connection closed" }
-
-            try {
-                callback(RawConnectionLease(conn))
-            } finally {
-                val updates = conn.readPendingUpdates()
-                if (updates.isNotEmpty()) {
-                    tableUpdatesFlow.emit(updates)
+            withContext(dispatcher) {
+                check(!closed) { "Connection closed" }
+
+                try {
+                    callback(RawConnectionLease(conn))
+                } finally {
+                    val updates = conn.readPendingUpdates()
+                    if (updates.isNotEmpty()) {
+                        tableUpdatesFlow.emit(updates)
+                    }
                 }
             }
         }
diff --git a/common/src/commonMain/kotlin/com/powersync/db/internal/InternalDatabaseImpl.kt b/common/src/commonMain/kotlin/com/powersync/db/internal/InternalDatabaseImpl.kt
@@ -11,26 +11,20 @@ import com.powersync.db.runWrapped
 import com.powersync.utils.AtomicMutableSet
 import com.powersync.utils.JsonUtil
 import com.powersync.utils.throttle
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.IO
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onSubscription
 import kotlinx.coroutines.flow.transform
-import kotlinx.coroutines.withContext
 import kotlin.time.Duration.Companion.milliseconds
 
 @OptIn(ExperimentalPowerSyncAPI::class)
 internal class InternalDatabaseImpl(
     private val pool: SQLiteConnectionPool,
     private val logger: Logger,
 ) : InternalDatabase {
-    // Could be scope.coroutineContext, but the default is GlobalScope, which seems like a bad idea. To discuss.
-    private val dbContext = Dispatchers.IO
-
     override suspend fun execute(
         sql: String,
         parameters: List<Any?>?,
@@ -40,20 +34,18 @@ internal class InternalDatabaseImpl(
         }
 
     override suspend fun updateSchema(schemaJson: String) {
-        withContext(dbContext) {
-            runWrapped {
-                pool.withAllConnections { writer, readers ->
-                    writer.runTransaction { tx ->
-                        tx.getOptional(
-                            "SELECT powersync_replace_schema(?);",
-                            listOf(schemaJson),
-                        ) {}
-                    }
+        runWrapped {
+            pool.withAllConnections { writer, readers ->
+                writer.runTransaction { tx ->
+                    tx.getOptional(
+                        "SELECT powersync_replace_schema(?);",
+                        listOf(schemaJson),
+                    ) {}
+                }
 
-                    // Update the schema on all read connections
-                    for (readConnection in readers) {
-                        readConnection.execSQL("pragma table_info('sqlite_master')")
-                    }
+                // Update the schema on all read connections
+                for (readConnection in readers) {
+                    readConnection.execSQL("pragma table_info('sqlite_master')")
                 }
             }
         }
@@ -167,11 +159,9 @@ internal class InternalDatabaseImpl(
      */
     @OptIn(ExperimentalPowerSyncAPI::class)
     private suspend fun <R> internalReadLock(callback: suspend (SQLiteConnectionLease) -> R): R =
-        withContext(dbContext) {
-            runWrapped {
-                useConnection(true) { connection ->
-                    callback(connection)
-                }
+        runWrapped {
+            useConnection(true) { connection ->
+                callback(connection)
             }
         }
 
@@ -189,11 +179,9 @@ internal class InternalDatabaseImpl(
 
     @OptIn(ExperimentalPowerSyncAPI::class)
     private suspend fun <R> internalWriteLock(callback: suspend (SQLiteConnectionLease) -> R): R =
-        withContext(dbContext) {
+        runWrapped {
             pool.write { writer ->
-                runWrapped {
-                    callback(writer)
-                }
+                callback(writer)
             }
         }
 
