Accumulate updated during throttle. Cleanup default throttle duration. Cleanup comments.

Author: stevensJourney

core/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt

@@ -350,7 +350,7 @@ internal class PowerSyncDatabaseImpl(
 
     override fun onChange(
         tables: Set<String>,
-        throttleMs: Long?,
+        throttleMs: Long,
     ): Flow<Set<String>> =
         flow {
             waitReady()
@@ -362,7 +362,7 @@ internal class PowerSyncDatabaseImpl(
     override fun <RowType : Any> watch(
         sql: String,
         parameters: List<Any?>?,
-        throttleMs: Long?,
+        throttleMs: Long,
         mapper: (SqlCursor) -> RowType,
     ): Flow<List<RowType>> =
         flow {

core/src/commonMain/kotlin/com/powersync/db/Queries.kt

@@ -5,6 +5,8 @@ import com.powersync.db.internal.ConnectionContext
 import com.powersync.db.internal.PowerSyncTransaction
 import kotlinx.coroutines.flow.Flow
 import kotlin.coroutines.cancellation.CancellationException
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.milliseconds
 
 public fun interface ThrowableTransactionCallback<R> {
     @Throws(PowerSyncException::class, kotlinx.coroutines.CancellationException::class)
@@ -17,6 +19,13 @@ public fun interface ThrowableLockCallback<R> {
 }
 
 public interface Queries {
+    public companion object {
+        /**
+         * The default throttle duration for  [onChange] and [watch] operations.
+         */
+        public val DEFAULT_THROTTLE: Duration = 30.milliseconds
+    }
+
     /**
      * Executes a write query (INSERT, UPDATE, DELETE).
      *
@@ -87,23 +96,23 @@ public interface Queries {
      * Returns a [Flow] that emits whenever the source tables are modified.
      *
      * @param tables The set of tables to monitor for changes.
-     * @param throttleMs The minimum interval, in milliseconds, between queries. Defaults to null.
+     * @param throttleMs The minimum interval, in milliseconds, between emissions. Defaults to [DEFAULT_THROTTLE]. Table changes are accumulated while throttling is active. The accumulated set of tables will be emitted on the trailing edge of the throttle.
      * @return A [Flow] emitting the set of modified tables.
      * @throws PowerSyncException If a database error occurs.
      * @throws CancellationException If the operation is cancelled.
      */
     @Throws(PowerSyncException::class, CancellationException::class)
     public fun onChange(
         tables: Set<String>,
-        throttleMs: Long? = null,
+        throttleMs: Long = DEFAULT_THROTTLE.inWholeMilliseconds,
     ): Flow<Set<String>>
 
     /**
      * Executes a read-only (SELECT) query every time the source tables are modified and returns the results as a [Flow] of lists.
      *
      * @param sql The SQL query to execute.
      * @param parameters The parameters for the query, or an empty list if none.
-     * @param throttleMs The minimum interval, in milliseconds, between queries. Defaults to null.
+     * @param throttleMs The minimum interval, in milliseconds, between queries. Defaults to [DEFAULT_THROTTLE].
      * @param mapper A function to map the result set to the desired type.
      * @return A [Flow] emitting lists of results.
      * @throws PowerSyncException If a database error occurs.
@@ -113,7 +122,7 @@ public interface Queries {
     public fun <RowType : Any> watch(
         sql: String,
         parameters: List<Any?>? = listOf(),
-        throttleMs: Long? = null,
+        throttleMs: Long = DEFAULT_THROTTLE.inWholeMilliseconds,
         mapper: (SqlCursor) -> RowType,
     ): Flow<List<RowType>>
 

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

@@ -8,6 +8,7 @@ import com.powersync.db.ThrowableLockCallback
 import com.powersync.db.ThrowableTransactionCallback
 import com.powersync.db.runWrapped
 import com.powersync.db.runWrappedSuspending
+import com.powersync.utils.AtomicMutableSet
 import com.powersync.utils.JsonUtil
 import com.powersync.utils.throttle
 import kotlinx.coroutines.CoroutineScope
@@ -56,10 +57,6 @@ internal class InternalDatabaseImpl(
     // Could be scope.coroutineContext, but the default is GlobalScope, which seems like a bad idea. To discuss.
     private val dbContext = Dispatchers.IO
 
-    companion object {
-        val DEFAULT_WATCH_THROTTLE = 30.milliseconds
-    }
-
     override suspend fun execute(
         sql: String,
         parameters: List<Any?>?,
@@ -108,36 +105,42 @@ internal class InternalDatabaseImpl(
 
     override fun onChange(
         tables: Set<String>,
-        throttleMs: Long?,
+        throttleMs: Long,
     ): 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>()
+
             updatesOnTables()
                 .transform { updates ->
                     val intersection = updates.intersect(watchedTables)
                     if (intersection.isNotEmpty()) {
                         // Transform table names using friendlyTableName
-                        emit(intersection.map { friendlyTableName(it) }.toSet())
+                        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 ?: DEFAULT_WATCH_THROTTLE)
+                .throttle(throttleMs.milliseconds)
                 .collect {
                     // Emit the transformed tables which have changed
-                    send(it)
+                    val copy = batchedUpdates.toSetAndClear()
+                    send(copy)
                 }
         }
 
     override fun <RowType : Any> watch(
         sql: String,
         parameters: List<Any?>?,
-        throttleMs: Long?,
+        throttleMs: Long,
         mapper: (SqlCursor) -> RowType,
     ): Flow<List<RowType>> =
         // Use a channel flow here since we throttle (buffer used under the hood)
@@ -163,7 +166,7 @@ internal class InternalDatabaseImpl(
                 // 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 ?: DEFAULT_WATCH_THROTTLE)
+                .throttle(throttleMs.milliseconds)
                 .collect {
                     send(getAll(sql, parameters = parameters, mapper = mapper))
                 }

core/src/commonMain/kotlin/com/powersync/utils/AtomicMutableSet.kt

@@ -11,6 +11,11 @@ public class AtomicMutableSet<T> : SynchronizedObject() {
             return set.add(element)
         }
 
+    public fun addAll(elements: Set<T>): Boolean =
+        synchronized(this) {
+            return set.addAll(elements)
+        }
+
     // Synchronized clear method
     public fun clear(): Unit =
         synchronized(this) {