From b96e9bae8e693fcf83459472cff2effc4b73bc91 Mon Sep 17 00:00:00 2001 From: LucDeCaf Date: Tue, 14 Jul 2026 16:40:40 +0200 Subject: [PATCH 1/2] Implement WatchManager --- .../Client/PowerSyncDatabase.cs | 325 ++------------ .../PowerSync.Common/Client/WatchManager.cs | 418 ++++++++++++++++++ .../PowerSync.Common/PowerSync.Common.csproj | 1 - 3 files changed, 446 insertions(+), 298 deletions(-) create mode 100644 PowerSync/PowerSync.Common/Client/WatchManager.cs diff --git a/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs b/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs index f14f1c5..11f75e1 100644 --- a/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs +++ b/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs @@ -1,16 +1,13 @@ namespace PowerSync.Common.Client; -using System.Runtime.CompilerServices; -using System.Text.RegularExpressions; -using System.Threading.Channels; using System.Threading.Tasks; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Newtonsoft.Json; + using Nito.AsyncEx; -using ThrottleDebounce; using PowerSync.Common.Client.Connection; using PowerSync.Common.Client.Sync.Bucket; @@ -132,9 +129,6 @@ public class PowerSyncDatabase : IPowerSyncDatabase public IDBAdapter Database { get; protected set; } private Schema schema; - 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; } public bool Ready { get; protected set; } @@ -145,6 +139,8 @@ public class PowerSyncDatabase : IPowerSyncDatabase private readonly InternalSubscriptionManager subscriptions; + private readonly WatchManager watchManager; + private StreamingSyncImplementation? syncStreamImplementation; public string SdkVersion { get; protected set; } @@ -204,6 +200,8 @@ public PowerSyncDatabase(PowerSyncDatabaseOptions options) remoteFactory = options.RemoteFactory ?? (connector => new Remote(connector)); + watchManager = new WatchManager(this, masterCts.Token); + // Start async init subscriptions = new InternalSubscriptionManager( firstStatusMatching: WaitForStatus, @@ -344,7 +342,9 @@ protected async Task Initialize(PowerSyncDatabaseOptions options) { await BucketStorageAdapter.Init(); await LoadVersion(); - await UpdateSchema(options.Schema); + // Note: no SchemaChangedEvent here - watched queries only resolve their source tables + // once initialization has completed, so the initial schema is never stale for them. + await ReplaceSchema(options.Schema); await ResolveOfflineSyncStatus(); await Database.Execute("PRAGMA RECURSIVE_TRIGGERS=TRUE"); Ready = true; @@ -407,6 +407,12 @@ public async Task UpdateSchema(Schema schema) throw new Exception("Cannot update schema while connected"); } + await ReplaceSchema(schema); + Events.Emit(new PowerSyncDBEvents.SchemaChangedEvent(schema)); + } + + private async Task ReplaceSchema(Schema schema) + { try { schema.Validate(); @@ -419,7 +425,6 @@ public async Task UpdateSchema(Schema schema) this.schema = schema; await Database.Execute("SELECT powersync_replace_schema(?)", [JsonConvert.SerializeObject(schema)]); await Database.RefreshSchema(); - Events.Emit(new PowerSyncDBEvents.SchemaChangedEvent(schema)); } /// @@ -768,192 +773,24 @@ public async Task WriteTransaction(Func> fn, DBLockO return await Database.WriteTransaction(fn, options); } - public IAsyncEnumerable OnChange(SQLWatchOptions? options = null) - { - options ??= new SQLWatchOptions(); - - var tables = options?.Tables ?? []; - var powersyncTables = new HashSet( - tables.SelectMany(table => new[] { $"ps_data__{table}", $"ps_data_local__{table}" }) - ); - - var signal = options?.Signal != null - ? CancellationTokenSource.CreateLinkedTokenSource(masterCts.Token, options.Signal.Value) - : CancellationTokenSource.CreateLinkedTokenSource(masterCts.Token); - - var listener = Database.Events.OnTablesUpdated.ListenAsync(signal.Token); - - // Return the actual IAsyncEnumerable here, using OnChange as a synchronous wrapper that blocks until the - // connection is established - var throttleMs = options?.ThrottleMs ?? DEFAULT_WATCH_THROTTLE_MS; - return OnChangeCore(powersyncTables, listener, signal, options?.TriggerImmediately == true, throttleMs); - } - - private async IAsyncEnumerable OnChangeCore( - HashSet watchedTables, - IAsyncEnumerable listener, - CancellationTokenSource signal, - bool triggerImmediately, - int throttleMs = DEFAULT_WATCH_THROTTLE_MS - ) - { - try - { - await foreach (var update in OnRawTableChange(watchedTables, listener, signal.Token, triggerImmediately, throttleMs)) - { - // Convert from 'ps_data__' to '' - for (int i = 0; i < update.ChangedTables.Length; i++) - { - update.ChangedTables[i] = InternalToFriendlyTableName(update.ChangedTables[i]); - } - yield return update; - } - } - finally - { - signal.Dispose(); - } - } - - private static string InternalToFriendlyTableName(string internalName) - { - const string PS_DATA_PREFIX = "ps_data__"; - const string PS_DATA_LOCAL_PREFIX = "ps_data_local__"; - - if (internalName.StartsWith(PS_DATA_PREFIX)) - return internalName.Substring(PS_DATA_PREFIX.Length); - - if (internalName.StartsWith(PS_DATA_LOCAL_PREFIX)) - return internalName.Substring(PS_DATA_LOCAL_PREFIX.Length); - - return internalName; - } - - public IAsyncEnumerable Watch( - string sql, - object?[]? parameters = null, - SQLWatchOptions? options = null - ) + /// + /// Executes a read query every time the source tables are modified. + /// + /// The query's source tables are resolved automatically (or taken from + /// when provided), and are re-resolved whenever the + /// schema changes, so watches keep working across calls. + /// + public IAsyncEnumerable Watch(string sql, object?[]? parameters = null, SQLWatchOptions? options = null) { - options ??= new SQLWatchOptions(); - - // Stop watching on master CTS cancellation, or on user CTS cancellation - var signal = options.Signal != null - ? CancellationTokenSource.CreateLinkedTokenSource(masterCts.Token, options.Signal.Value) - : CancellationTokenSource.CreateLinkedTokenSource(masterCts.Token); - - // Establish the initial DB listener synchronously before returning the IAsyncEnumerable, - // so that table changes between Watch() being called and iteration starting are not missed. - // This mirrors the pattern used in OnChange(). - var initialRestartCts = CancellationTokenSource.CreateLinkedTokenSource(signal.Token); - var initialListener = Database.Events.OnTablesUpdated.ListenAsync(initialRestartCts.Token); - - return WatchCore(sql, parameters, options, signal, initialRestartCts, initialListener); + return watchManager.Watch(sql, parameters, options); } - private async IAsyncEnumerable WatchCore( - string sql, - object?[]? parameters, - SQLWatchOptions options, - CancellationTokenSource signal, - CancellationTokenSource initialRestartCts, - IAsyncEnumerable initialListener - ) + /// + /// Emits an event whenever any of the tables in are modified. + /// + public IAsyncEnumerable OnChange(SQLWatchOptions? options = null) { - var schemaChanged = new TaskCompletionSource(); - - // Listen for schema changes in the background - var schemaListenerTask = Task.Run(async () => - { - await foreach (var update in Events.OnSchemaChanged.ListenAsync(signal.Token)) - { - // Swap schemaChanged with an unresolved TCS - var oldTcs = Interlocked.Exchange(ref schemaChanged, new()); - oldTcs.TrySetResult(true); - } - }, signal.Token); - - // Re-register query on schema updates - bool isRestart = false; - var currentRestartCts = initialRestartCts; - var currentListener = initialListener; - var throttleMs = options?.ThrottleMs ?? DEFAULT_WATCH_THROTTLE_MS; - - try - { - while (!signal.Token.IsCancellationRequested) - { - // Resolve tables - HashSet powersyncTables; - if (options?.Tables != null) - { - powersyncTables = [.. options - .Tables - .SelectMany(table => [$"ps_data__{table}", $"ps_data_local__{table}"] - )]; - } - else - { - powersyncTables = await GetSourceTables(sql, parameters); - } - - var enumerator = OnRawTableChange( - powersyncTables, - currentListener, - currentRestartCts.Token, - isRestart || (options?.TriggerImmediately == true), - throttleMs - ).GetAsyncEnumerator(); - - // Continually wait for either OnChange or SchemaChanged to fire - while (true) - { - var currentSchemaTask = schemaChanged.Task; - var onChangeTask = enumerator.MoveNextAsync().AsTask(); - var completedTask = await Task.WhenAny(onChangeTask, currentSchemaTask); - - if (completedTask == currentSchemaTask) - { - var oldRestartCts = currentRestartCts; - oldRestartCts.Cancel(); - isRestart = true; - // Let the current task complete/cancel gracefully - try { await onChangeTask; } - catch (OperationCanceledException) { } - - // Establish a new listener BEFORE resolving source tables in the next iteration, - // so that changes during the async GetSourceTables call are not missed. - currentRestartCts = CancellationTokenSource.CreateLinkedTokenSource(signal.Token); - currentListener = Database.Events.OnTablesUpdated.ListenAsync(currentRestartCts.Token); - oldRestartCts.Dispose(); - - break; - } - - // Await onChangeTask to propagate cancellation and detect end-of-enumeration - bool hasNext; - try { hasNext = await onChangeTask; } - catch (OperationCanceledException) { yield break; } - - if (!hasNext) break; - - var update = enumerator.Current; - if (update.ChangedTables != null) - { - yield return await GetAll(sql, parameters); - } - } - } - } - finally - { - signal.Cancel(); - try { await schemaListenerTask; } - catch (OperationCanceledException) { } - - currentRestartCts.Dispose(); - signal.Dispose(); - } + return watchManager.OnChange(options); } private class ExplainedResult @@ -985,112 +822,6 @@ internal async Task> GetSourceTables(string sql, object?[]? para return [.. tables.Select(row => row.tbl_name)]; } - - private async IAsyncEnumerable OnRawTableChange( - HashSet watchedTables, - IAsyncEnumerable listener, - [EnumeratorCancellation] CancellationToken signal, - bool triggerImmediately = false, - int throttleMs = DEFAULT_WATCH_THROTTLE_MS - ) - { - if (triggerImmediately) - { - yield return new WatchOnChangeEvent { ChangedTables = [] }; - } - - if (throttleMs <= 0) - { - // No throttling - HashSet changedTables = new(); - await foreach (var e in listener) - { - GetTablesFromNotification(e.TablesUpdated, changedTables); - changedTables.IntersectWith(watchedTables); - if (changedTables.Count == 0) continue; - yield return new WatchOnChangeEvent { ChangedTables = [.. changedTables] }; - } - yield break; - } - - // Throttled - publish via throttled call to an action that flushes accumulated changes into this channel - var channel = Channel.CreateUnbounded(); - var accumulatedTables = new HashSet(); - - _ = Task.Run(async () => - { - using var throttledFlush = Throttler.Throttle(() => - { - // Safe to lock directly on accumulatedTables because it's a local variable - lock (accumulatedTables) - { - if (accumulatedTables.Count == 0) return; - channel.Writer.TryWrite(new WatchOnChangeEvent { ChangedTables = [.. accumulatedTables] }); - accumulatedTables.Clear(); - } - }, - TimeSpan.FromMilliseconds(throttleMs), - leading: false, - trailing: true - ); - - try - { - var changedTables = new HashSet(); - await foreach (var e in listener) - { - GetTablesFromNotification(e.TablesUpdated, changedTables); - changedTables.IntersectWith(watchedTables); - if (changedTables.Count == 0) continue; - - lock (accumulatedTables) { accumulatedTables.UnionWith(changedTables); } - throttledFlush.Invoke(); - } - } - catch (OperationCanceledException) { } - finally - { - // Flush any remaining events and close the channel - lock (accumulatedTables) - { - if (accumulatedTables.Count > 0) - { - channel.Writer.TryWrite(new WatchOnChangeEvent { ChangedTables = [.. accumulatedTables] }); - accumulatedTables.Clear(); - } - } - channel.Writer.Complete(); - } - }); - - // Continuously pull values from channel and publish to the consumer - while (await channel.Reader.WaitToReadAsync(CancellationToken.None)) - { - while (channel.Reader.TryRead(out var evt)) - { - yield return evt; - } - } - } - - private static void GetTablesFromNotification(INotification updateNotification, HashSet changedTables) - { - changedTables.Clear(); - string[] tables = []; - if (updateNotification is BatchedUpdateNotification batchedUpdate) - { - tables = batchedUpdate.Tables; - } - else if (updateNotification is UpdateNotification singleUpdate) - { - tables = [singleUpdate.Table]; - } - - foreach (var table in tables) - { - changedTables.Add(table); - } - } } public class SQLWatchOptions diff --git a/PowerSync/PowerSync.Common/Client/WatchManager.cs b/PowerSync/PowerSync.Common/Client/WatchManager.cs new file mode 100644 index 0000000..3e09665 --- /dev/null +++ b/PowerSync/PowerSync.Common/Client/WatchManager.cs @@ -0,0 +1,418 @@ +namespace PowerSync.Common.Client; + +using System.Collections.Concurrent; +using System.Threading.Channels; + +using Microsoft.Extensions.Logging; + +using PowerSync.Common.DB; + +/// +/// Owns all watched queries for a . +/// +/// A single listener on the DB adapter's table-update events dispatches changes to each +/// registered , and a single listener on schema changes marks +/// every subscription's resolved source tables as stale. Subscriptions are otherwise fully +/// independent: each one throttles, re-resolves its source tables and re-runs its query on +/// its own consumer-driven loop, so a slow or failing query only affects its own stream. +/// +internal class WatchManager +{ + internal const int DEFAULT_THROTTLE_MS = 30; + + private const string PS_DATA_PREFIX = "ps_data__"; + private const string PS_DATA_LOCAL_PREFIX = "ps_data_local__"; + + private readonly PowerSyncDatabase db; + private readonly CancellationToken masterToken; + private readonly ConcurrentDictionary watches = new(); + + public WatchManager(PowerSyncDatabase db, CancellationToken masterToken) + { + this.db = db; + this.masterToken = masterToken; + + // Dispatch two tasks, one for TableUpdated events and one for SchemaChanged events + _ = Task.Run(RunTableUpdateLoop); + _ = Task.Run(RunSchemaChangeLoop); + } + + public IAsyncEnumerable Watch(string sql, object?[]? parameters, SQLWatchOptions? options) + { + options ??= new SQLWatchOptions(); + + // Register synchronously so that table changes between this call and the consumer + // starting iteration are not missed. + var subscription = CreateSubscription( + options, + resolveTables: options.Tables != null + ? () => Task.FromResult(ExpandTableNames(options.Tables)) + : () => db.GetSourceTables(sql, parameters), + initialTables: options.Tables != null ? ExpandTableNames(options.Tables) : null, + // Don't flush updates mid-cancellation, since that may involve expensive + // database operations. + flushOnCancel: false + ); + + return Stream(subscription, _ => db.GetAll(sql, parameters)); + } + + public IAsyncEnumerable OnChange(SQLWatchOptions? options) + { + options ??= new SQLWatchOptions(); + + var tables = ExpandTableNames(options.Tables ?? []); + var subscription = CreateSubscription( + options, + resolveTables: () => Task.FromResult(new HashSet(tables)), + initialTables: tables, + // Deliver changes accumulated during the throttle window even if cancellation + // lands before the window expires. + flushOnCancel: true + ); + + return Stream(subscription, changed => Task.FromResult(new WatchOnChangeEvent + { + ChangedTables = [.. changed.Select(InternalToFriendlyTableName)] + })); + } + + private WatchSubscription CreateSubscription( + SQLWatchOptions options, + Func>> resolveTables, + HashSet? initialTables, + bool flushOnCancel + ) + { + var cts = options.Signal != null + ? CancellationTokenSource.CreateLinkedTokenSource(masterToken, options.Signal.Value) + : CancellationTokenSource.CreateLinkedTokenSource(masterToken); + + var subscription = new WatchSubscription( + resolveTables, + initialTables, + options.ThrottleMs ?? DEFAULT_THROTTLE_MS, + options.TriggerImmediately, + flushOnCancel, + cts + ); + + watches.TryAdd(subscription, 0); + + // Cleanup must not depend on the consumer ever iterating the stream: on any cancellation + // (external Signal, master token, or Unregister) drop the subscription from the registry. + // CTS disposal stays in Unregister - disposing from inside a cancellation callback is unsafe. + cts.Token.Register(() => watches.TryRemove(subscription, out _)); + + return subscription; + } + + private void Unregister(WatchSubscription subscription) + { + if (!subscription.TryClose()) return; + + watches.TryRemove(subscription, out _); + subscription.Cts.Cancel(); + subscription.Cts.Dispose(); + } + + private async IAsyncEnumerable Stream(WatchSubscription subscription, Func, Task> onChange) + { + try + { + while (true) + { + var changed = await subscription.WaitForChangeAsync(); + if (changed == null) yield break; + + yield return await onChange(changed); + } + } + finally + { + Unregister(subscription); + } + } + + private async Task RunTableUpdateLoop() + { + try + { + await foreach (var update in db.Database.Events.OnTablesUpdated.ListenAsync(masterToken)) + { + // Prevent a single bad notification from taking down the entire TablesUpdated loop + try + { + var changed = new HashSet(DBAdapterUtils.ExtractTableUpdates(update.TablesUpdated)); + if (changed.Count == 0) continue; + + foreach (var entry in watches) + { + entry.Key.NotifyTablesChanged(changed); + } + } + catch (Exception ex) + { + db.Logger.LogError(ex, "Failed to dispatch a table update to watched queries."); + } + } + } + catch (OperationCanceledException) { } + catch (Exception ex) + { + db.Logger.LogError(ex, "Watch OnTablesUpdated dispatcher failed; terminating all watched queries."); + } + finally + { + foreach (var entry in watches) + { + watches.TryRemove(entry.Key, out _); + entry.Key.Terminate(); + } + } + } + + private async Task RunSchemaChangeLoop() + { + try + { + await foreach (var _ in db.Events.OnSchemaChanged.ListenAsync(masterToken)) + { + foreach (var entry in watches) + { + entry.Key.RequestRefresh(); + } + } + } + catch (OperationCanceledException) { } + catch (Exception ex) + { + db.Logger.LogError(ex, "Watch OnSchemaChanged dispatcher failed; watched queries will not react to schema changes."); + } + } + + private static HashSet ExpandTableNames(IEnumerable tables) => + [.. tables.SelectMany(table => new[] { $"{PS_DATA_PREFIX}{table}", $"{PS_DATA_LOCAL_PREFIX}{table}" })]; + + private static string InternalToFriendlyTableName(string internalName) + { + if (internalName.StartsWith(PS_DATA_PREFIX)) + return internalName.Substring(PS_DATA_PREFIX.Length); + + if (internalName.StartsWith(PS_DATA_LOCAL_PREFIX)) + return internalName.Substring(PS_DATA_LOCAL_PREFIX.Length); + + return internalName; + } +} + +/// +/// Dispatch state for a single watched query. +/// +/// The manager pushes changed table names in via and schema +/// resets via ; the consumer pulls coalesced change batches out via +/// . Throttling is per-subscription: the signal channel has a +/// capacity of one, so any number of notifications during the throttle window or an in-flight +/// query collapse into a single pending wake-up. +/// +internal class WatchSubscription +{ + private readonly object mutex = new(); + private readonly Channel signal = Channel.CreateBounded( + new BoundedChannelOptions(1) { FullMode = BoundedChannelFullMode.DropWrite } + ); + + private readonly Func>> resolveTables; + private readonly int throttleMs; + private readonly bool flushOnCancel; + // Static table sets (explicit SQLWatchOptions.Tables) survive schema refreshes: only the + // underlying data changes, never the expansion, so a refresh re-runs the query without + // discarding the set. + private readonly bool staticTables; + + private int closed; + + // The resolved source tables of the query. Null means "not resolved" - either not yet + // resolved, or invalidated by a schema change. While null, all table notifications are + // accumulated unfiltered and filtered against the newly resolved set later, so no + // relevant change is missed during (re)resolution. + private HashSet? tables; + private HashSet pending = []; + private bool refreshRequested; + + internal CancellationTokenSource Cts { get; } + + public WatchSubscription( + Func>> resolveTables, + HashSet? initialTables, + int throttleMs, + bool triggerImmediately, + bool flushOnCancel, + CancellationTokenSource cts + ) + { + this.resolveTables = resolveTables; + this.throttleMs = throttleMs; + this.flushOnCancel = flushOnCancel; + staticTables = initialTables != null; + tables = initialTables; + Cts = cts; + + if (triggerImmediately) + { + refreshRequested = true; + signal.Writer.TryWrite(true); + } + } + + public void NotifyTablesChanged(HashSet changed) + { + lock (mutex) + { + if (tables == null) + { + pending.UnionWith(changed); + } + else + { + var relevant = false; + foreach (var table in changed) + { + if (tables.Contains(table)) + { + pending.Add(table); + relevant = true; + } + } + if (!relevant) return; + } + } + signal.Writer.TryWrite(true); + } + + public void RequestRefresh() + { + lock (mutex) + { + refreshRequested = true; + if (!staticTables) tables = null; + } + signal.Writer.TryWrite(true); + } + + /// + /// Ends the subscription's stream gracefully without cancelling it: any pending batch is + /// still delivered, then returns null. Used when the + /// dispatcher can no longer deliver updates. + /// + public void Terminate() + { + signal.Writer.TryComplete(); + } + + /// + /// Marks the subscription closed. Returns false if it was already closed, making + /// unregistration idempotent across multiple enumerations of the same stream. + /// + public bool TryClose() + { + return Interlocked.Exchange(ref closed, 1) == 0; + } + + /// + /// Waits for the next batch of relevant table changes, applying the throttle window. + /// Returns the changed table names (empty for refresh/immediate triggers, where the query + /// must run regardless of which tables changed), or null when the subscription ends. + /// + public async Task?> WaitForChangeAsync() + { + while (true) + { + try + { + if (!await signal.Reader.WaitToReadAsync(Cts.Token)) return null; + } + catch (OperationCanceledException) + { + return FlushPendingOnCancel(); + } + catch (ObjectDisposedException) + { + // A prior enumeration of the same stream already unregistered this subscription + return null; + } + + bool refresh; + lock (mutex) refresh = refreshRequested; + + // Refresh and TriggerImmediately wake-ups bypass the throttle so that consumers see + // post-schema-change results (and initial results) without delay. + if (!refresh && throttleMs > 0) + { + try + { + await Task.Delay(throttleMs, Cts.Token); + } + catch (OperationCanceledException) + { + return FlushPendingOnCancel(); + } + catch (ObjectDisposedException) + { + return null; + } + } + + // Consume the signal only after the throttle window, so notifications that arrived + // during the window coalesce into this batch instead of causing another wake-up. + signal.Reader.TryRead(out _); + + HashSet batch; + HashSet? currentTables; + lock (mutex) + { + refresh = refreshRequested; + refreshRequested = false; + batch = pending; + pending = []; + currentTables = tables; + } + + if (currentTables == null) + { + currentTables = await resolveTables(); + lock (mutex) + { + // A concurrent RequestRefresh may have nulled `tables` again; don't clobber it, + // the next wake-up will re-resolve. + if (!refreshRequested) tables = currentTables; + } + } + + // Names accumulated while unresolved are unfiltered; restrict them to the actual + // source tables. Already-filtered names are unaffected. + batch.IntersectWith(currentTables); + + if (refresh) return batch; + if (batch.Count == 0) continue; + return batch; + } + } + + private HashSet? FlushPendingOnCancel() + { + if (!flushOnCancel) return null; + + lock (mutex) + { + if (tables == null) return null; + + pending.IntersectWith(tables); + if (pending.Count == 0) return null; + + var batch = pending; + pending = []; + return batch; + } + } +} diff --git a/PowerSync/PowerSync.Common/PowerSync.Common.csproj b/PowerSync/PowerSync.Common/PowerSync.Common.csproj index bc69111..6e45bb7 100644 --- a/PowerSync/PowerSync.Common/PowerSync.Common.csproj +++ b/PowerSync/PowerSync.Common/PowerSync.Common.csproj @@ -35,7 +35,6 @@ - From 66e9865bc2877285547b940bf677c476b967d70e Mon Sep 17 00:00:00 2001 From: LucDeCaf Date: Wed, 15 Jul 2026 10:53:06 +0200 Subject: [PATCH 2/2] Await db init in Watch_SchemaReset --- .../PowerSync.Common.Tests/Client/PowerSyncDatabaseTests.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/PowerSyncDatabaseTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/PowerSyncDatabaseTests.cs index 6af182e..eec7820 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/PowerSyncDatabaseTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/PowerSyncDatabaseTests.cs @@ -678,6 +678,7 @@ public async Task Watch_SchemaReset() }, Schema = TestSchema.MakeOptionalSyncSchema(false) }); + await db.Init(); try {