Lock MDSQLiteConnection._updateBuffer before accessing it

PowerSync/PowerSync.Common/CHANGELOG.md

@@ -5,6 +5,7 @@
 - **Breaking:** Made `Table.Name` non-nullable (default ""). This change may affect 0% of users, but it is technically a breaking change.
 - Add support for loading custom SQLite extensions via `MDSQLiteOptions.Extensions`.
 - Fix streaming sync retry loop reconnecting with no delay after an exception, ignoring `RetryDelayMs`.
+- Lock `MDSQLiteConnection._updateBuffer` to fix a rare issue with high-frequency local writes while syncing.
 - (internal) Remove `Compiled*` classes in favor of working with `Table` and `Schema` objects directly.
 
 ## 0.1.2

PowerSync/PowerSync.Common/MDSQLite/MDSQLiteConnection.cs

@@ -21,19 +21,23 @@ public class MDSQLiteConnectionOptions(SqliteConnection database)
 public class MDSQLiteConnection : EventStream<DBAdapterEvents.TablesUpdatedEvent>, ILockContext
 {
     public SqliteConnection Db;
-    private readonly List<UpdateNotification> updateBuffer;
+    private readonly List<UpdateNotification> _updateBuffer;
+    private readonly object _updateBufferLock = new();
     public MDSQLiteConnection(MDSQLiteConnectionOptions options)
     {
         Db = options.Database;
-        updateBuffer = [];
+        _updateBuffer = [];
 
         raw.sqlite3_rollback_hook(Db.Handle, RollbackHook, IntPtr.Zero);
         raw.sqlite3_update_hook(Db.Handle, UpdateHook, IntPtr.Zero);
     }
 
     private void RollbackHook(object user_data)
     {
-        updateBuffer.Clear();
+        lock (_updateBufferLock)
+        {
+            _updateBuffer.Clear();
+        }
     }
 
     private void UpdateHook(object user_data, int type, utf8z database, utf8z table, long rowId)
@@ -45,17 +49,26 @@ private void UpdateHook(object user_data, int type, utf8z database, utf8z table,
             23 => RowUpdateType.SQLITE_UPDATE,
             _ => throw new InvalidOperationException($"Unknown update type: {type}"),
         };
-        updateBuffer.Add(new UpdateNotification(table.utf8_to_string(), opType, rowId));
+        lock (_updateBufferLock)
+        {
+            _updateBuffer.Add(new UpdateNotification(table.utf8_to_string(), opType, rowId));
+        }
     }
 
     public void FlushUpdates()
     {
-        if (updateBuffer.Count == 0)
+        UpdateNotification[] updates;
+        lock (_updateBufferLock)
         {
-            return;
+            if (_updateBuffer.Count == 0)
+            {
+                return;
+            }
+            updates = _updateBuffer.ToArray();
+            _updateBuffer.Clear();
         }
 
-        var groupedUpdates = updateBuffer
+        var groupedUpdates = updates
             .GroupBy(update => update.Table)
             .ToDictionary(
                 group => group.Key,
@@ -65,11 +78,10 @@ public void FlushUpdates()
         var batchedUpdate = new BatchedUpdateNotification
         {
             GroupedUpdates = groupedUpdates,
-            RawUpdates = updateBuffer.ToArray(),
+            RawUpdates = updates.ToArray(),
             Tables = groupedUpdates.Keys.ToArray()
         };
 
-        updateBuffer.Clear();
         Emit(new DBAdapterEvents.TablesUpdatedEvent(batchedUpdate));
     }