diff --git a/PowerSync/PowerSync.Common/CHANGELOG.md b/PowerSync/PowerSync.Common/CHANGELOG.md index e1726cf..130cbd9 100644 --- a/PowerSync/PowerSync.Common/CHANGELOG.md +++ b/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 diff --git a/PowerSync/PowerSync.Common/MDSQLite/MDSQLiteConnection.cs b/PowerSync/PowerSync.Common/MDSQLite/MDSQLiteConnection.cs index f2b9003..6a612bd 100644 --- a/PowerSync/PowerSync.Common/MDSQLite/MDSQLiteConnection.cs +++ b/PowerSync/PowerSync.Common/MDSQLite/MDSQLiteConnection.cs @@ -21,11 +21,12 @@ public class MDSQLiteConnectionOptions(SqliteConnection database) public class MDSQLiteConnection : EventStream, ILockContext { public SqliteConnection Db; - private readonly List updateBuffer; + private readonly List _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); @@ -33,7 +34,10 @@ public MDSQLiteConnection(MDSQLiteConnectionOptions options) 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)); }