Throttle Watch<T> and OnChange

LucDeCaf · LucDeCaf · commit f3dc694bfa06 · 2026-03-05T11:18:58.000+02:00
diff --git a/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs b/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs
@@ -1,5 +1,6 @@
 namespace PowerSync.Common.Client;
 
+using System.Diagnostics;
 using System.Runtime.CompilerServices;
 using System.Text.RegularExpressions;
 using System.Threading.Tasks;
@@ -131,7 +132,7 @@ public class PowerSyncDatabase : IPowerSyncDatabase
     public IDBAdapter Database { get; protected set; }
     private CompiledSchema schema;
 
-    private static readonly int DEFAULT_WATCH_THROTTLE_MS = 30;
+    private const int DEFAULT_WATCH_THROTTLE_MS = 30;
     private static readonly Regex POWERSYNC_TABLE_MATCH = new Regex(@"(^ps_data__|^ps_data_local__)", RegexOptions.Compiled);
 
     public bool Closed { get; protected set; }
@@ -785,19 +786,21 @@ public IAsyncEnumerable<WatchOnChangeEvent> OnChange(SQLWatchOptions? options =
 
         // Return the actual IAsyncEnumerable here, using OnChange as a synchronous wrapper that blocks until the
         // connection is established
-        return OnChangeCore(powersyncTables, listener, signal, options?.TriggerImmediately == true);
+        var throttleMs = options?.ThrottleMs ?? DEFAULT_WATCH_THROTTLE_MS;
+        return OnChangeCore(powersyncTables, listener, signal, options?.TriggerImmediately == true, throttleMs);
     }
 
     private async IAsyncEnumerable<WatchOnChangeEvent> OnChangeCore(
         HashSet<string> watchedTables,
         IAsyncEnumerable<DBAdapterEvents.TablesUpdatedEvent> listener,
         CancellationTokenSource signal,
-        bool triggerImmediately
+        bool triggerImmediately,
+        int throttleMs = DEFAULT_WATCH_THROTTLE_MS
     )
     {
         try
         {
-            await foreach (var update in OnRawTableChange(watchedTables, listener, signal.Token, triggerImmediately))
+            await foreach (var update in OnRawTableChange(watchedTables, listener, signal.Token, triggerImmediately, throttleMs))
             {
                 // Convert from 'ps_data__<name>' to '<name>'
                 for (int i = 0; i < update.ChangedTables.Length; i++)
@@ -875,6 +878,7 @@ private async IAsyncEnumerable<T[]> WatchCore<T>(
         bool isRestart = false;
         var currentRestartCts = initialRestartCts;
         var currentListener = initialListener;
+        var throttleMs = options?.ThrottleMs ?? DEFAULT_WATCH_THROTTLE_MS;
 
         try
         {
@@ -898,7 +902,8 @@ private async IAsyncEnumerable<T[]> WatchCore<T>(
                     powersyncTables,
                     currentListener,
                     currentRestartCts.Token,
-                    isRestart || (options?.TriggerImmediately == true)
+                    isRestart || (options?.TriggerImmediately == true),
+                    throttleMs
                 ).GetAsyncEnumerator();
 
                 // Continually wait for either OnChange or SchemaChanged to fire
@@ -986,26 +991,111 @@ private async IAsyncEnumerable<WatchOnChangeEvent> OnRawTableChange(
         HashSet<string> watchedTables,
         IAsyncEnumerable<DBAdapterEvents.TablesUpdatedEvent> listener,
         [EnumeratorCancellation] CancellationToken token,
-        bool triggerImmediately = false
+        bool triggerImmediately = false,
+        int throttleMs = DEFAULT_WATCH_THROTTLE_MS
     )
     {
         if (triggerImmediately)
         {
             yield return new WatchOnChangeEvent { ChangedTables = [] };
         }
 
-        HashSet<string> changedTables = new();
-        await foreach (var e in listener)
+        if (throttleMs <= 0)
         {
-            // Extract the changed tables and intersect with the watched tables
-            changedTables.Clear();
-            GetTablesFromNotification(e.TablesUpdated, changedTables);
-            changedTables.IntersectWith(watchedTables);
+            // Fast path: no throttling
+            HashSet<string> changedTables = new();
+            await foreach (var e in listener)
+            {
+                changedTables.Clear();
+                GetTablesFromNotification(e.TablesUpdated, changedTables);
+                changedTables.IntersectWith(watchedTables);
+                if (changedTables.Count == 0) continue;
+                yield return new WatchOnChangeEvent { ChangedTables = [.. changedTables] };
+            }
+            yield break;
+        }
+
+        // Throttled path: Task.WhenAny loop with leading-edge emit
+        var listenerEnumerator = listener.GetAsyncEnumerator(token);
+        try
+        {
+            var accumulated = new HashSet<string>();
+            long lastYieldTime = 0;
+            Task<bool> moveNextTask = listenerEnumerator.MoveNextAsync().AsTask();
+            Task? throttleTask = null;
+
+            while (true)
+            {
+                if (throttleTask != null)
+                    await Task.WhenAny(moveNextTask, throttleTask);
+                else
+                {
+                    try { await moveNextTask; }
+                    catch (OperationCanceledException) { break; }
+                }
 
-            if (changedTables.Count == 0) continue;
+                // Throttle timer fired without a new event
+                if (throttleTask != null && throttleTask.IsCompleted && !moveNextTask.IsCompleted)
+                {
+                    if (accumulated.Count > 0)
+                    {
+                        lastYieldTime = Stopwatch.GetTimestamp();
+                        yield return new WatchOnChangeEvent { ChangedTables = [.. accumulated] };
+                        accumulated.Clear();
+                    }
+                    throttleTask = null;
+                    continue;
+                }
+
+                // A new event arrived (possibly alongside throttle)
+                bool hasNext;
+                try { hasNext = await moveNextTask; }
+                catch (OperationCanceledException) { break; }
+                if (!hasNext) break;
 
-            yield return new WatchOnChangeEvent { ChangedTables = [.. changedTables] };
+                AccumulateMatchingTables(listenerEnumerator.Current, watchedTables, accumulated);
+
+                if (accumulated.Count > 0)
+                {
+                    var now = Stopwatch.GetTimestamp();
+                    var elapsedMs = (now - lastYieldTime) * 1000.0 / Stopwatch.Frequency;
+
+                    if (elapsedMs >= throttleMs)
+                    {
+                        // Leading edge: emit immediately
+                        lastYieldTime = now;
+                        yield return new WatchOnChangeEvent { ChangedTables = [.. accumulated] };
+                        accumulated.Clear();
+                        throttleTask = null;
+                    }
+                    else
+                    {
+                        throttleTask ??= Task.Delay((int)(throttleMs - elapsedMs), token);
+                    }
+                }
+
+                moveNextTask = listenerEnumerator.MoveNextAsync().AsTask();
+            }
+
+            if (accumulated.Count > 0)
+                yield return new WatchOnChangeEvent { ChangedTables = [.. accumulated] };
         }
+        finally
+        {
+            await listenerEnumerator.DisposeAsync();
+        }
+    }
+
+    private static void AccumulateMatchingTables(
+        DBAdapterEvents.TablesUpdatedEvent e,
+        HashSet<string> watchedTables,
+        HashSet<string> accumulated
+    )
+    {
+        var tables = new HashSet<string>();
+        GetTablesFromNotification(e.TablesUpdated, tables);
+        tables.IntersectWith(watchedTables);
+        accumulated.UnionWith(tables);
     }
 
     private static void GetTablesFromNotification(INotification updateNotification, HashSet<string> changedTables)
diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/PowerSyncDatabaseTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/PowerSyncDatabaseTests.cs
@@ -842,26 +842,196 @@ public async Task Watch_TriggerImmediately_False()
     [Fact(Timeout = 2000)]
     public async Task Watch_CancelsOnTokenCancellation()
     {
-        var cts = CancellationTokenSource.CreateLinkedTokenSource(testCts.Token);
         var tcs = new TaskCompletionSource<bool>();
-        var sem = new SemaphoreSlim(0);
+
         var listener = db.Watch<CountResult>(
             "SELECT COUNT(*) AS count FROM assets",
             null,
-            new() { Signal = cts.Token, TriggerImmediately = false });
+            new() { Signal = testCts.Token });
 
         // Sem == received result
         // TCS == received cancellation
         _ = Task.Run(async () =>
         {
-            await foreach (var _ in listener) { sem.Release(); }
+            await foreach (var _ in listener) { }
             tcs.TrySetResult(true);
         });
 
         await TestUtils.InsertRandomAssets(db, 3);
-        Assert.True(await sem.WaitAsync(200));
 
-        cts.Cancel();
+        testCts.Cancel();
+        Assert.True(await tcs.Task);
+    }
+
+    [Fact(Timeout = 5000)]
+    public async Task OnChange_ThrottlesBatchesRapidChanges()
+    {
+        int eventCount = 0;
+        var tcs = new TaskCompletionSource<bool>();
+
+        var listener = db.OnChange(new SQLWatchOptions
+        {
+            Tables = ["assets"],
+            Signal = testCts.Token,
+            ThrottleMs = 200,
+        });
+
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                await foreach (var _ in listener)
+                {
+                    Interlocked.Increment(ref eventCount);
+                }
+                tcs.TrySetResult(true);
+            }
+            catch (Exception ex)
+            {
+                tcs.TrySetException(ex);
+            }
+        });
+
+        for (int i = 0; i < 5; i++)
+        {
+            await TestUtils.InsertRandomAsset(db);
+        }
+
+        testCts.Cancel();
+        Assert.True(await tcs.Task);
+
+        Assert.True(eventCount < 5, $"Expected fewer than 5 events but got {eventCount}");
+    }
+
+    [Fact(Timeout = 5000)]
+    public async Task Watch_ThrottlesBatchesRapidChanges()
+    {
+        int eventCount = 0;
+        long lastCount = 0;
+        var tcs = new TaskCompletionSource<bool>();
+
+        var listener = db.Watch<CountResult>(
+            "SELECT COUNT(*) AS count FROM assets",
+            null,
+            new() { Signal = testCts.Token, ThrottleMs = 200 });
+
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                await foreach (var rows in listener)
+                {
+                    lastCount = rows.First().count;
+                    Interlocked.Increment(ref eventCount);
+                }
+                tcs.TrySetResult(true);
+            }
+            catch (Exception ex)
+            {
+                tcs.TrySetException(ex);
+            }
+        });
+
+        for (int i = 0; i < 5; i++)
+        {
+            await TestUtils.InsertRandomAsset(db);
+        }
+
+        testCts.Cancel();
+        Assert.True(await tcs.Task);
+
+        Assert.Equal(5, lastCount);
+        Assert.True(eventCount < 5, $"Expected fewer than 5 events but got {eventCount}");
+    }
+
+    [Fact(Timeout = 5000)]
+    public async Task OnChange_NoThrottleWhenZero()
+    {
+        int eventCount = 0;
+        using var sem = new SemaphoreSlim(0);
+
+        var listener = db.OnChange(new SQLWatchOptions
+        {
+            Tables = ["assets"],
+            Signal = testCts.Token,
+            ThrottleMs = 0,
+        });
+
+        _ = Task.Run(async () =>
+        {
+            await foreach (var _ in listener)
+            {
+                Interlocked.Increment(ref eventCount);
+                sem.Release();
+            }
+        }, testCts.Token);
+
+        for (int i = 0; i < 5; i++)
+        {
+            await db.Execute("INSERT INTO assets(id, description) VALUES(?, ?)", [Guid.NewGuid().ToString(), "test"]);
+            Assert.True(await sem.WaitAsync(500));
+        }
+
+        Assert.Equal(5, eventCount);
+    }
+
+    [Fact(Timeout = 5000)]
+    public async Task OnChange_FirstChangeIsNotDelayed()
+    {
+        using var sem = new SemaphoreSlim(0);
+        var sw = Stopwatch.StartNew();
+
+        var listener = db.OnChange(new SQLWatchOptions
+        {
+            Tables = ["assets"],
+            Signal = testCts.Token,
+            ThrottleMs = 500,
+        });
+
+        _ = Task.Run(async () =>
+        {
+            await foreach (var _ in listener)
+            {
+                sem.Release();
+                break;
+            }
+        }, testCts.Token);
+
+        await db.Execute("INSERT INTO assets(id, description) VALUES(?, ?)", [Guid.NewGuid().ToString(), "test"]);
+
+        Assert.True(await sem.WaitAsync(2000));
+        Assert.True(sw.ElapsedMilliseconds < 200, $"First event took {sw.ElapsedMilliseconds}ms, expected <200ms");
+    }
+
+    [Fact(Timeout = 5000)]
+    public async Task OnChange_ThrottleCancelledCleanly()
+    {
+        var tcs = new TaskCompletionSource<bool>();
+
+        var listener = db.OnChange(new SQLWatchOptions
+        {
+            Tables = ["assets"],
+            Signal = testCts.Token,
+            ThrottleMs = 500,
+        });
+
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                await foreach (var _ in listener) { }
+                tcs.TrySetResult(true);
+            }
+            catch (Exception ex)
+            {
+                tcs.TrySetException(ex);
+            }
+        });
+
+        // Insert to trigger the throttle delay, then cancel before the window expires
+        await db.Execute("INSERT INTO assets(id, description) VALUES(?, ?)", [Guid.NewGuid().ToString(), "test"]);
+        await Task.Delay(50);
+        testCts.Cancel();
 
         Assert.True(await tcs.Task);
     }
diff --git a/Tests/PowerSync/PowerSync.Common.Tests/PowerSync.Common.Tests.csproj b/Tests/PowerSync/PowerSync.Common.Tests/PowerSync.Common.Tests.csproj
@@ -32,8 +32,4 @@
     <ProjectReference Include="..\..\..\PowerSync\PowerSync.Common\PowerSync.Common.csproj" />
   </ItemGroup>
 
-  <ItemGroup>
-    <Content Include="xunit.runner.json" CopyToOutputDirectory="PreserveNewest" />
-  </ItemGroup>
-
 </Project>
diff --git a/Tests/PowerSync/PowerSync.Common.Tests/xunit.runner.json b/Tests/PowerSync/PowerSync.Common.Tests/xunit.runner.json
