Make watch and onChange more reliable

simolus3 · simolus3 · commit 6121ea1c1481 · 2025-09-16T16:44:06.000+02:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -5,6 +5,7 @@
 * Remove internal SQLDelight and SQLiter dependencies.
 * Add `rawConnection` getter to `ConnectionContext`, which is a `SQLiteConnection` instance from
   `androidx.sqlite` that can be used to step through statements in a custom way.
+* Fix an issue where `watch()` would run queries more often than intended.
 
 ## 1.5.1
 
diff --git a/core/src/commonIntegrationTest/kotlin/com/powersync/DatabaseTest.kt b/core/src/commonIntegrationTest/kotlin/com/powersync/DatabaseTest.kt
@@ -211,11 +211,10 @@ class DatabaseTest {
 
                 var changeSet = query.awaitItem()
                 // The initial result
-                changeSet.count() shouldBe 0
+                changeSet shouldHaveSize 0
 
                 changeSet = query.awaitItem()
-                changeSet.count() shouldBe 1
-                changeSet.contains("users") shouldBe true
+                changeSet shouldBe setOf("users")
 
                 query.cancel()
             }
diff --git a/core/src/commonMain/kotlin/com/powersync/db/internal/InternalDatabaseImpl.kt b/core/src/commonMain/kotlin/com/powersync/db/internal/InternalDatabaseImpl.kt
@@ -1,7 +1,6 @@
 package com.powersync.db.internal
 
 import com.powersync.ExperimentalPowerSyncAPI
-import com.powersync.PowerSyncException
 import com.powersync.db.SqlCursor
 import com.powersync.db.ThrowableLockCallback
 import com.powersync.db.ThrowableTransactionCallback
@@ -15,8 +14,9 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.IO
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.SharedFlow
-import kotlinx.coroutines.flow.channelFlow
-import kotlinx.coroutines.flow.filter
+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
@@ -79,75 +79,69 @@ internal class InternalDatabaseImpl(
         tables: Set<String>,
         throttleMs: Long,
         triggerImmediately: Boolean,
-    ): Flow<Set<String>> =
-        channelFlow {
-            // Match all possible internal table combinations
-            val watchedTables =
-                tables.flatMap { listOf(it, "ps_data__$it", "ps_data_local__$it") }.toSet()
-
-            // Accumulate updates between throttles
-            val batchedUpdates = AtomicMutableSet<String>()
+    ): Flow<Set<String>> {
+        // Match all possible internal table combinations
+        val watchedTables =
+            tables.flatMap { listOf(it, "ps_data__$it", "ps_data_local__$it") }.toSet()
 
-            updatesOnTables()
-                .onSubscription {
-                    if (triggerImmediately) {
-                        // Emit an initial event (if requested). No changes would be detected at this point
-                        send(setOf())
-                    }
-                }.transform { updates ->
-                    val intersection = updates.intersect(watchedTables)
-                    if (intersection.isNotEmpty()) {
-                        // Transform table names using friendlyTableName
-                        val friendlyTableNames = intersection.map { friendlyTableName(it) }.toSet()
-                        batchedUpdates.addAll(friendlyTableNames)
-                        emit(Unit)
-                    }
-                }
-                // Throttling here is a feature which prevents watch queries from spamming updates.
-                // Throttling by design discards and delays events within the throttle window. Discarded events
-                // still trigger a trailing edge update.
-                // Backpressure is avoided on the throttling and consumer level by buffering the last upstream value.
-                .throttle(throttleMs.milliseconds)
-                .collect {
-                    // Emit the transformed tables which have changed
-                    val copy = batchedUpdates.toSetAndClear()
-                    send(copy)
-                }
+        return rawChangedTables(watchedTables, throttleMs, triggerImmediately).map {
+            it.mapTo(mutableSetOf(), ::friendlyTableName)
         }
+    }
 
     override fun <RowType : Any> watch(
         sql: String,
         parameters: List<Any?>?,
         throttleMs: Long,
         mapper: (SqlCursor) -> RowType,
     ): Flow<List<RowType>> =
-        // Use a channel flow here since we throttle (buffer used under the hood)
-        // This causes some emissions to be from different scopes.
-        channelFlow {
+        flow {
             // Fetch the tables asynchronously with getAll
             val tables =
                 getSourceTables(sql, parameters)
                     .filter { it.isNotBlank() }
                     .toSet()
 
+            val queries =
+                rawChangedTables(tables, throttleMs, triggerImmediately = true).map {
+                    getAll(sql, parameters = parameters, mapper = mapper)
+                }
+            emitAll(queries)
+        }
+
+    private fun rawChangedTables(
+        tableNames: Set<String>,
+        throttleMs: Long,
+        triggerImmediately: Boolean,
+    ): Flow<Set<String>> =
+        flow {
+            val batchedUpdates = AtomicMutableSet<String>()
+
             updatesOnTables()
-                // onSubscription here is very important.
-                // This ensures that the initial result and all updates are emitted.
                 .onSubscription {
-                    send(getAll(sql, parameters = parameters, mapper = mapper))
-                }.filter {
-                    // Only trigger updates on relevant tables
-                    it.intersect(tables).isNotEmpty()
+                    if (triggerImmediately) {
+                        // Emit an initial event (if requested). No changes would be detected at this point
+                        emit(initialUpdateSentinel)
+                    }
+                }.transform { updates ->
+                    if (updates === initialUpdateSentinel) {
+                        // This should always be emitted despite being empty and not intersecting with
+                        // the tables we care about.
+                        emit(Unit)
+                    } else {
+                        val intersection = updates.intersect(tableNames)
+                        if (intersection.isNotEmpty()) {
+                            batchedUpdates.addAll(intersection)
+                            emit(Unit)
+                        }
+                    }
                 }
                 // Throttling here is a feature which prevents watch queries from spamming updates.
                 // Throttling by design discards and delays events within the throttle window. Discarded events
                 // still trigger a trailing edge update.
                 // Backpressure is avoided on the throttling and consumer level by buffering the last upstream value.
-                // Note that the buffered upstream "value" only serves to trigger the getAll query. We don't buffer watch results.
                 .throttle(throttleMs.milliseconds)
-                .collect {
-                    send(getAll(sql, parameters = parameters, mapper = mapper))
-                }
+                .collect { emit(batchedUpdates.toSetAndClear()) }
         }
 
     override suspend fun <T> useConnection(
@@ -259,6 +253,10 @@ internal class InternalDatabaseImpl(
         val p2: Long,
         val p3: Long,
     )
+
+    private companion object {
+        val initialUpdateSentinel = emptySet<String>()
+    }
 }
 
 /**
diff --git a/core/src/commonMain/kotlin/com/powersync/utils/AtomicMutableSet.kt b/core/src/commonMain/kotlin/com/powersync/utils/AtomicMutableSet.kt
@@ -11,7 +11,7 @@ public class AtomicMutableSet<T> : SynchronizedObject() {
             return set.add(element)
         }
 
-    public fun addAll(elements: Set<T>): Boolean =
+    public fun addAll(elements: Collection<T>): Boolean =
         synchronized(this) {
             return set.addAll(elements)
         }
diff --git a/core/src/commonMain/kotlin/com/powersync/utils/ThrottleFlow.kt b/core/src/commonMain/kotlin/com/powersync/utils/ThrottleFlow.kt
@@ -6,25 +6,34 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.flow
 import kotlin.time.Duration
+import kotlin.time.TimeSource
 
 /**
- * Throttles a flow with emissions on the leading and trailing edge.
- * Events, from the incoming flow, during the throttle window are discarded.
- * Events are discarded by using a conflated buffer.
- * This throttle method acts as a slow consumer, but backpressure is not a concern
- * due to the conflated buffer dropping events during the throttle window.
+ * Throttles an upstream flow.
+ *
+ * When a new event is emitted on this (upstream) flow, it is passed on downstream. For each value
+ * passed downstream, the resulting flow will pause for at least [window] (or longer if emitting
+ * the value downstream takes longer).
+ *
+ * While this flow is paused, no further events are passed downstream. The latest upstream event
+ * emitted during the pause state is buffered and handled once the pause is over.
+ *
+ * In other words, this flow will _drop events_, so it should only be used when the upstream flow
+ * serves as a notification marker (meaning that something downstream needs to run in response to
+ * events, but the actual event does not matter).
  */
 internal fun <T> Flow<T>.throttle(window: Duration): Flow<T> =
     flow {
         // Use a buffer before throttle (ensure only the latest event is kept)
         val bufferedFlow = this@throttle.buffer(Channel.CONFLATED)
 
         bufferedFlow.collect { value ->
-            // Emit the event immediately (leading edge)
+            // Pause for the downstream emit or the delay window, whatever is longer
+            val pauseUntil = TimeSource.Monotonic.markNow() + window
             emit(value)
 
-            // Delay for the throttle window to avoid emitting too frequently
-            delay(window)
+            // Negating the duration because we want to pause until pauseUntil has passed.
+            delay(-pauseUntil.elapsedNow())
 
             // The next incoming event will be provided from the buffer.
             // The next collect will emit the trailing edge
diff --git a/core/src/commonTest/kotlin/com/powersync/utils/JsonTest.kt b/core/src/commonTest/kotlin/com/powersync/utils/JsonTest.kt
@@ -1,10 +1,5 @@
 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
@@ -23,7 +18,6 @@ import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
-import kotlin.time.Duration.Companion.milliseconds
 
 class JsonTest {
     @Test
@@ -42,28 +36,6 @@ 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.milliseconds)
-                    .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)
diff --git a/core/src/commonTest/kotlin/com/powersync/utils/ThrottleTest.kt b/core/src/commonTest/kotlin/com/powersync/utils/ThrottleTest.kt
@@ -0,0 +1,61 @@
+package com.powersync.utils
+
+import io.kotest.matchers.shouldBe
+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 kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.seconds
+
+class ThrottleTest {
+    @Test
+    fun testThrottle() {
+        runTest {
+            val t =
+                flow {
+                    emit(1)
+                    delay(10)
+                    emit(2)
+                    delay(20)
+                    emit(3)
+                    delay(100)
+                    emit(4)
+                }.throttle(100.milliseconds)
+                    .map {
+                        // Adding a delay here to simulate a slow consumer
+                        delay(1000)
+                        it
+                    }.toList()
+            assertEquals(t, listOf(1, 4))
+        }
+    }
+
+    @Test
+    fun testWaitTimeIsNotAdditive() =
+        runTest {
+            val upstream =
+                flow {
+                    repeat(5) {
+                        emit(it)
+                        delay(10.seconds)
+                    }
+                }
+
+            // If throttle were to start the delay after the downstream emit completed, it would 11
+            // seconds, skipping events. That's not what we want though, the downstream delay is
+            // long enough to not need additional throttling.
+            val events =
+                upstream
+                    .throttle(5.seconds)
+                    .map {
+                        delay(6.seconds)
+                        it
+                    }.toList()
+
+            events shouldBe listOf(0, 1, 2, 3, 4)
+        }
+}

Original file line number	Diff line number	Diff line change
`@@ -11,7 +11,7 @@ public class AtomicMutableSet<T> : SynchronizedObject() {`
`11`	`11`	`return set.add(element)`
`12`	`12`	`}`
`13`	`13`
`14`		`- public fun addAll(elements: Set<T>): Boolean =`
	`14`	`+ public fun addAll(elements: Collection<T>): Boolean =`
`15`	`15`	`synchronized(this) {`
`16`	`16`	`return set.addAll(elements)`
`17`	`17`	`}`