fix race conditions

Author: stevensJourney

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

@@ -27,20 +27,18 @@ internal class PsSqlDriver(
     }
 
     fun clearTableUpdates() {
-        // This should only ever be executed on rollback which should be executed via the
-        // IO Dispatcher.
+        // This should only ever be executed by an execute operation which should
+        // always be executed with the IO Dispatcher
         runBlocking {
             pendingUpdates.clear()
         }
     }
 
     // Flows on any table change
-    // This specifically returns a SharedFlow for timing considerations
-    fun updatesOnTables(): SharedFlow<Set<String>> {
-        // Spread the input table names in order to account for internal views
-        return tableUpdatesFlow
+    // This specifically returns a SharedFlow for downstream timing considerations
+    fun updatesOnTables(): SharedFlow<Set<String>> =
+        tableUpdatesFlow
             .asSharedFlow()
-    }
 
     suspend fun fireTableUpdates() {
         val updates = pendingUpdates.toSet(true)

core/src/commonMain/kotlin/com/powersync/db/internal/InternalDatabaseImpl.kt

@@ -166,25 +166,16 @@ internal class InternalDatabaseImpl(
                     .filter { it.isNotBlank() }
                     .toSet()
 
-            // Register a listener before fetching the initial result
-            val updateFlow =
-                updatesOnTables()
-
-            val initialResult = getAll(sql, parameters = parameters, mapper = mapper)
-            // Listen for updates before emitting the initial result
-
-            updateFlow
+            updatesOnTables()
                 // onSubscription here is very important.
                 // This ensures that the initial result and all updates are emitted.
                 .onSubscription {
-                    println("emitting initial result")
-                    send(initialResult)
+                    send(getAll(sql, parameters = parameters, mapper = mapper))
                 }.filter {
                     // Only trigger updates on relevant tables
                     it.intersect(tables).isNotEmpty()
                 }.throttle(throttleMs ?: DEFAULT_WATCH_THROTTLE_MS)
                 .collect {
-                    println("mapping update to result")
                     send(getAll(sql, parameters = parameters, mapper = mapper))
                 }
         }

core/src/commonTest/kotlin/com/powersync/utils/JsonTest.kt

@@ -1,5 +1,10 @@
 package com.powersync.utils
 
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.test.runTest
 import kotlinx.serialization.json.JsonArray
 import kotlinx.serialization.json.JsonNull
 import kotlinx.serialization.json.JsonObject
@@ -36,6 +41,28 @@ class JsonTest {
         assertEquals("test", jsonElement.content)
     }
 
+    @Test
+    fun testThrottle() {
+        runTest {
+            val t =
+                flow {
+                    emit(1)
+                    delay(10)
+                    emit(2)
+                    delay(20)
+                    emit(3)
+                    delay(100)
+                    emit(4)
+                }.throttle(100)
+                    .map {
+                        // Adding a delay here to simulate a slow consumer
+                        delay(1000)
+                        it
+                    }.toList()
+            assertEquals(t, listOf(1, 4))
+        }
+    }
+
     @Test
     fun testBooleanToJsonElement() {
         val boolean = JsonParam.Boolean(true)