even more fixes

Author: simolus3

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

@@ -29,6 +29,7 @@ import dev.mokkery.verifySuspend
 import io.kotest.matchers.shouldBe
 import io.kotest.matchers.shouldNotBe
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.onEach
 import kotlinx.io.files.Path
@@ -60,7 +61,17 @@ class AttachmentsTest {
         database
             .watch("SELECT * FROM attachments") {
                 Attachment.fromCursor(it)
-            }.onEach { logger.i { "attachments table results: $it" } }
+            }
+            // Because tests run on slow machines, it's possible for a schedule like the following
+            // to happen:
+            //  1. the attachment is initially saved with QUEUED_DOWNLOAD, triggering a query.
+            //  2. the attachment is downloaded fast, but the query flow is paused.
+            //  3. only now is the query scheduled by 1 actually running, reporting SYNCED.
+            //  4. we delete the attachment.
+            //  5. thanks to 2, the query runs again, again reporting SYNCED.
+            //  6. Our test now fails because the second event should be ARCHIVED.
+            .distinctUntilChanged()
+            .onEach { logger.i { "attachments table results: $it" } }
 
     suspend fun updateSchema(db: PowerSyncDatabase) {
         db.updateSchema(

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

@@ -112,6 +112,8 @@ internal class ActiveDatabaseTest(
                 dbDirectory = testDirectory,
                 logger = logger,
                 scope = scope,
+                // We want to reduce concurrency in tests to make their output more predictable.
+                readPoolSize = 1,
             )
         doOnCleanup { db.close() }
         return db

core/src/commonMain/kotlin/com/powersync/PowerSyncDatabaseFactory.kt

@@ -40,6 +40,7 @@ public fun PowerSyncDatabase(
         scope = scope,
         logger = generatedLogger,
         dbDirectory = dbDirectory,
+        readPoolSize = 5,
     )
 }
 
@@ -51,6 +52,7 @@ internal fun createPowerSyncDatabaseImpl(
     scope: CoroutineScope,
     logger: Logger,
     dbDirectory: String?,
+    readPoolSize: Int,
 ): PowerSyncDatabaseImpl {
     val identifier = dbDirectory + dbFilename
     val activeDatabaseGroup = ActiveDatabaseGroup.referenceDatabase(logger, identifier)
@@ -63,6 +65,7 @@ internal fun createPowerSyncDatabaseImpl(
                 dbFilename,
                 dbDirectory,
                 activeDatabaseGroup.first.group.writeLockMutex,
+                readPoolSize = readPoolSize,
             )
         }
 

core/src/commonMain/kotlin/com/powersync/db/driver/InternalConnectionPool.kt

@@ -20,9 +20,10 @@ internal class InternalConnectionPool(
     private val dbFilename: String,
     private val dbDirectory: String?,
     private val writeLockMutex: Mutex,
+    readPoolSize: Int,
 ) : SQLiteConnectionPool {
     private val writeConnection = newConnection(false)
-    private val readPool = ReadPool({ newConnection(true) }, scope = scope)
+    private val readPool = ReadPool({ newConnection(true) }, size = readPoolSize, scope = scope)
 
     // MutableSharedFlow to emit batched table updates
     private val tableUpdatesFlow = MutableSharedFlow<Set<String>>(replay = 0)

core/src/commonMain/kotlin/com/powersync/db/driver/ReadPool.kt

@@ -18,7 +18,7 @@ import kotlinx.coroutines.launch
 @OptIn(ExperimentalPowerSyncAPI::class)
 internal class ReadPool(
     factory: () -> SQLiteConnection,
-    size: Int = 5,
+    size: Int,
     private val scope: CoroutineScope,
 ) {
     private val available = Channel<Pair<SQLiteConnection, CompletableDeferred<Unit>>>()