Better cleanup in tests, fix issues with orphaned tasks / disposal order

Author: LucDeCaf

PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs

@@ -286,18 +286,22 @@ public async Task WaitForStatus(Func<SyncStatus, bool> predicate, CancellationTo
         }
 
         var tcs = new TaskCompletionSource<bool>();
-        var cts = new CancellationTokenSource();
+        var cts = CancellationTokenSource.CreateLinkedTokenSource(masterCts.Token);
 
-        _ = Task.Run(() =>
+        _ = Task.Run(async () =>
         {
-            foreach (var update in Listen(cts.Token))
+            try
             {
-                if (update.StatusChanged != null && predicate(update.StatusChanged))
+                await foreach (var update in ListenAsync(cts.Token))
                 {
-                    cts.Cancel();
-                    tcs.TrySetResult(true);
+                    if (update.StatusChanged != null && predicate(update.StatusChanged))
+                    {
+                        cts.Cancel();
+                        tcs.TrySetResult(true);
+                    }
                 }
             }
+            catch (OperationCanceledException) { }
         });
 
         cancellationToken?.Register(() =>

PowerSync/PowerSync.Common/MDSQLite/MDSQLiteAdapter.cs

@@ -31,6 +31,7 @@ public class MDSQLiteAdapter : EventStream<DBAdapterEvent>, IDBAdapter
 
     protected RequiredMDSQLiteOptions resolvedMDSQLiteOptions;
     private CancellationTokenSource? tablesUpdatedCts;
+    private Task? tablesUpdatedTask;
 
     private readonly AsyncLock writeMutex = new();
     private readonly AsyncLock readMutex = new();
@@ -86,11 +87,7 @@ private async Task Init()
 
         foreach (var statement in writeConnectionStatements)
         {
-            for (int tries = 0; tries < 30; tries++)
-            {
-                await writeConnection!.Execute(statement);
-                tries = 30;
-            }
+            await writeConnection!.Execute(statement);
         }
 
         foreach (var statement in readConnectionStatements)
@@ -99,9 +96,9 @@ private async Task Init()
         }
 
         tablesUpdatedCts = new CancellationTokenSource();
-        var _ = Task.Run(() =>
+        tablesUpdatedTask = Task.Run(async () =>
         {
-            foreach (var notification in writeConnection!.Listen(tablesUpdatedCts.Token))
+            await foreach (var notification in writeConnection!.ListenAsync(tablesUpdatedCts.Token))
             {
                 if (notification.TablesUpdated != null)
                 {
@@ -139,6 +136,7 @@ protected virtual void LoadExtension(SqliteConnection db)
     public new void Close()
     {
         tablesUpdatedCts?.Cancel();
+        try { tablesUpdatedTask?.Wait(TimeSpan.FromSeconds(2)); } catch { }
         base.Close();
         writeConnection?.Close();
         readConnection?.Close();

PowerSync/PowerSync.Common/MDSQLite/MDSQLiteConnection.cs

@@ -212,7 +212,6 @@ public async Task<NonQueryResult> Execute(string query, object?[]? parameters =
         DynamicParameters? dynamicParams = PrepareQuery(ref query, parameters);
         int rowsAffected = await Db.ExecuteAsync(query, dynamicParams, commandType: CommandType.Text);
 
-        FlushUpdates();
         return new NonQueryResult
         {
             InsertId = raw.sqlite3_last_insert_rowid(Db.Handle),

Tests/PowerSync/PowerSync.Common.Tests/Client/PowerSyncDatabaseTests.cs

@@ -6,6 +6,7 @@ namespace PowerSync.Common.Tests.Client;
 
 using PowerSync.Common.Client;
 using PowerSync.Common.Tests.Models;
+using PowerSync.Common.Tests.Utils;
 
 /// <summary>
 /// dotnet test -v n --framework net8.0 --filter "PowerSyncDatabaseTests"
@@ -678,96 +679,114 @@ await db.Execute(
     public async Task WatchSchemaResetTest()
     {
         var dbId = Guid.NewGuid().ToString();
+        var localDbName = $"powerSyncWatch_{dbId}.db";
         var db = new PowerSyncDatabase(new()
         {
             Database = new SQLOpenOptions
             {
-                DbFilename = $"powerSyncWatch_{dbId}.db",
+                DbFilename = localDbName,
             },
             Schema = TestSchema.MakeOptionalSyncSchema(false)
         });
 
-        using var sem = new SemaphoreSlim(0);
-        long lastCount = 0;
-
-        const string QUERY = "SELECT COUNT(*) AS count FROM assets";
-        var listener = db.Watch<CountResult>(QUERY, null, new() { Signal = testCts.Token });
-        _ = Task.Run(async () =>
+        try
         {
-            await foreach (var result in listener)
+            using var sem = new SemaphoreSlim(0);
+            long lastCount = 0;
+
+            const string QUERY = "SELECT COUNT(*) AS count FROM assets";
+            var listener = db.Watch<CountResult>(QUERY, null, new() { Signal = testCts.Token });
+            _ = Task.Run(async () =>
             {
-                if (result.Length > 0)
+                await foreach (var result in listener)
                 {
-                    lastCount = result[0].count;
+                    if (result.Length > 0)
+                    {
+                        lastCount = result[0].count;
+                    }
+                    sem.Release();
                 }
-                sem.Release();
-            }
-        });
+            });
 
-        var resolved = await db.GetSourceTables(QUERY, null);
-        Assert.Single(resolved);
-        Assert.Contains("ps_data_local__local_assets", resolved);
+            var resolved = await db.GetSourceTables(QUERY, null);
+            Assert.Single(resolved);
+            Assert.Contains("ps_data_local__local_assets", resolved);
 
-        for (int i = 0; i < 3; i++)
-        {
-            await db.Execute(
-                "insert into assets(id, description, make) values (?, ?, ?)",
-                [Guid.NewGuid().ToString(), "some desc", "some make"]
-            );
-            Assert.True(await sem.WaitAsync(100));
-            Assert.Equal(i + 1, lastCount);
-        }
-        Assert.Equal(3, lastCount);
+            for (int i = 0; i < 3; i++)
+            {
+                await db.Execute(
+                    "insert into assets(id, description, make) values (?, ?, ?)",
+                    [Guid.NewGuid().ToString(), "some desc", "some make"]
+                );
+                Assert.True(await sem.WaitAsync(100));
+                Assert.Equal(i + 1, lastCount);
+            }
+            Assert.Equal(3, lastCount);
 
-        await db.UpdateSchema(TestSchema.MakeOptionalSyncSchema(true));
+            await db.UpdateSchema(TestSchema.MakeOptionalSyncSchema(true));
 
-        resolved = await db.GetSourceTables(QUERY);
-        Assert.Single(resolved);
-        Assert.Contains("ps_data__assets", resolved);
+            resolved = await db.GetSourceTables(QUERY);
+            Assert.Single(resolved);
+            Assert.Contains("ps_data__assets", resolved);
 
-        Assert.True(await sem.WaitAsync(100));
-        Assert.Equal(0, lastCount);
+            Assert.True(await sem.WaitAsync(100));
+            Assert.Equal(0, lastCount);
 
-        await db.Execute("insert into assets select * from inactive_local_assets");
-        Assert.True(await sem.WaitAsync(500));
-        Assert.Equal(3, lastCount);
+            await db.Execute("insert into assets select * from inactive_local_assets");
+            Assert.True(await sem.WaitAsync(500));
+            Assert.Equal(3, lastCount);
 
-        // Sanity check
-        testCts.Cancel();
-        await Task.Delay(100);
+            // Sanity check
+            testCts.Cancel();
+            await Task.Delay(100);
 
-        await db.Execute("delete from assets");
-        Assert.False(await sem.WaitAsync(100));
-        Assert.Equal(3, lastCount);
+            await db.Execute("delete from assets");
+            Assert.False(await sem.WaitAsync(100));
+            Assert.Equal(3, lastCount);
+        }
+        finally
+        {
+            await db.Close();
+            DatabaseUtils.CleanDb(localDbName);
+        }
     }
 
     [Fact]
     public async Task Attributes_ColumnAliasing()
     {
+        var localDbName = $"PowerSyncAttributesTest-{Guid.NewGuid():N}.db";
         var db = new PowerSyncDatabase(new PowerSyncDatabaseOptions
         {
-            Database = new SQLOpenOptions { DbFilename = "PowerSyncAttributesTest.db" },
+            Database = new SQLOpenOptions { DbFilename = localDbName },
             Schema = TestSchemaAttributes.AppSchema,
         });
-        await db.DisconnectAndClear();
+        try
+        {
+            await db.DisconnectAndClear();
 
-        var id = Guid.NewGuid().ToString();
-        var description = "Test description";
-        var completed = false;
-        var createdAt = DateTimeOffset.Now;
+            var id = Guid.NewGuid().ToString();
+            var description = "Test description";
+            var completed = false;
+            var createdAt = DateTimeOffset.Now;
 
-        await db.Execute(
-            "INSERT INTO todos(id, description, completed, created_at, list_id) VALUES(?, ?, ?, ?, uuid())",
-            [id, description, completed, createdAt]
-        );
+            await db.Execute(
+                "INSERT INTO todos(id, description, completed, created_at, list_id) VALUES(?, ?, ?, ?, uuid())",
+                [id, description, completed, createdAt]
+            );
 
-        var results = await db.GetAll<Todo>("SELECT * FROM todos");
-        Assert.Single(results);
-        var row = results.First();
-        Assert.Equal(id, row.TodoId);
-        Assert.Equal(description, row.Description);
-        Assert.Equal(completed, row.Completed);
-        Assert.Equal(createdAt, row.CreatedAt);
+            var results = await db.GetAll<Todo>("SELECT * FROM todos");
+            Assert.Single(results);
+            var row = results.First();
+            Assert.Equal(id, row.TodoId);
+            Assert.Equal(description, row.Description);
+            Assert.Equal(completed, row.Completed);
+            Assert.Equal(createdAt, row.CreatedAt);
+        }
+        finally
+        {
+            await db.Close();
+            DatabaseUtils.CleanDb(localDbName);
+        }
     }
 
     [Fact]