Passing null payloads to control (if exists). Invoke notifyCompletedUploads after crud upload completes.

Author: Chriztiaan

PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs

@@ -443,7 +443,7 @@ public async Task<string> Control(string op, object? payload = null)
     {
         return await db.WriteTransaction(async tx =>
         {
-            var result = await tx.Get<ControlResult>("SELECT powersync_control(?, ?) AS r", [op, payload ?? ""]);
+            var result = await tx.Get<ControlResult>("SELECT powersync_control(?, ?) AS r", [op, payload]);
             return result.r!;
         });
     }

PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs

@@ -141,6 +141,8 @@ public class StreamingSyncImplementation : EventStream<StreamingSyncImplementati
     private Task? streamingSyncTask;
     public Action TriggerCrudUpload { get; }
     private CancellationTokenSource? crudUpdateCts;
+
+    private bool isUploadingCrud;
     private Action? notifyCompletedUploads; private readonly ILogger logger;
 
     private readonly StreamingSyncLocks locks;
@@ -162,19 +164,24 @@ public StreamingSyncImplementation(StreamingSyncImplementationOptions options)
 
         locks = new StreamingSyncLocks();
         logger = options.Logger ?? NullLogger.Instance;
+        isUploadingCrud = false;
 
         CancellationTokenSource = null;
 
         TriggerCrudUpload = () =>
         {
-            if (!SyncStatus.Connected)
+            if (!SyncStatus.Connected || isUploadingCrud)
             {
                 return;
             }
 
-            notifyCompletedUploads?.Invoke();
-
-            Task.Run(async () => await InternalUploadAllCrud());
+            isUploadingCrud = true;
+            Task.Run(async () =>
+            {
+                await InternalUploadAllCrud();
+                notifyCompletedUploads?.Invoke();
+                isUploadingCrud = false;
+            });
         };
     }
 

PowerSync/PowerSync.Common/DB/Crud/SyncStatus.cs

@@ -33,7 +33,7 @@ public class SyncDataFlowStatus
 
 public class SyncPriorityStatus
 {
-    [JsonProperty("uploading")] public int Priority { get; set; }
+    [JsonProperty("priority")] public int Priority { get; set; }
 
     [JsonProperty("lastSyncedAt")] public DateTime? LastSyncedAt { get; set; }
 

Tests/PowerSync/PowerSync.Common.IntegrationTests/SyncIntegrationTests.cs

@@ -49,7 +49,7 @@ public async Task InitializeAsync()
         {
             await db.Connect(connector, new PowerSyncConnectionOptions
             {
-                ClientImplementation = SyncClientImplementation.C_SHARP,
+                ClientImplementation = SyncClientImplementation.RUST,
             });
             await db.WaitForFirstSync();
         }