New checkpoint request protocol

Directory.build.props

@@ -2,7 +2,6 @@
 <Project>
     <PropertyGroup>
         <MSBuildWarningsAsMessages>$(MSBuildWarningsAsMessages);NETSDK1202</MSBuildWarningsAsMessages>
-        <GenerateDocumentationFile>true</GenerateDocumentationFile>
     </PropertyGroup>
     <ItemGroup>
         <Compile Include="$(MSBuildThisFileDirectory)IsExternalInit.cs" Visible="false" />

PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs

@@ -25,3 +25,28 @@ public interface IPowerSyncBackendConnector
     /// </summary>
     Task UploadData(IPowerSyncDatabase database);
 }
+
+/// <summary>
+/// An <see cref="IPowerSyncBackendConnector" /> capable of requesting checkpoints.
+///
+/// Extend this class instead of <see cref="IPowerSyncBackendConnector" /> when uploads are processed
+/// asynchronously by the backend (for example through a message queue): The sync client as part of
+/// the PowerSync .NET SDK generates a checkpoint request id and hands it to your backend via this
+/// class, which is responsible for creating a matching checkpoint once the uploads preceding the
+/// request have been processed.
+/// For more details, see <see href="https://docs.powersync.com/client-sdks/advanced/checkpoint-requests#asynchronous-upload-backends">asynchronous backend uploads</see>.
+///
+/// To use this connector, using <see cref="Sync.Stream.CheckpointMode.Requests" /> is required. Note that
+/// this requires PowerSync service version 1.24.0 or later.
+/// </summary>
+public interface ICustomCheckpointRequestConnector : IPowerSyncBackendConnector
+{
+    /// <summary>
+    /// Posts a client-generated checkpoint request to the backend and returns the effective checkpoint request state.
+    /// <para />
+    /// Currently, checkpoint request IDs are represented as strings. This is because some PowerSync SDKs are for runtimes
+    /// that don't have a fast 64-bit integer type. In a future release, checkpoint request IDs will change to be
+    /// represented by longs, meaning the <paramref name="requestId" /> parameter's type will also change to `long`.
+    /// </summary>
+    Task<string> PostCheckpointRequest(string clientId, string requestId);
+}

PowerSync/PowerSync.Common/Client/ConnectionManager.cs

@@ -175,6 +175,11 @@ public async Task Connect(IPowerSyncBackendConnector connector, PowerSyncConnect
         // Update pending options to the latest values
         PendingConnectionOptions = new StoredConnectionOptions(connector, options);
 
+        // Warn if connector expects checkpoint requests but write checkpoints are enabled
+        if (connector is ICustomCheckpointRequestConnector && PendingConnectionOptions.Options.CheckpointMode == CheckpointMode.Legacy)
+        {
+            Logger.LogWarning("The backend connector implements ICustomCheckpointRequestConnector, but Connect() was called without checkpoint requests enabled.");
+        }
 
         // Disconnecting here provides aborting in progress connection attempts.
         // The ConnectInternal method will clear pending options once it starts connecting (with the options).

PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs

@@ -50,6 +50,13 @@ public class PowerSyncDatabaseOptions() : BasePowerSyncDatabaseOptions()
     /// If not provided, a default Remote will be created.
     /// </summary>
     public Func<IPowerSyncBackendConnector, Remote>? RemoteFactory { get; set; }
+
+    /// <summary>
+    /// Source of the delays used by the sync client (retry delays, upload throttling).
+    /// Defaults to <see cref="System.TimeProvider.System" />; tests substitute a fake clock so they
+    /// don't have to wait out real retry delays.
+    /// </summary>
+    internal TimeProvider? TimeProvider { get; set; }
 }
 
 public class PowerSyncDBEvents : EventManager
@@ -200,14 +207,15 @@ public PowerSyncDatabase(PowerSyncDatabaseOptions options)
         SdkVersion = "";
 
         remoteFactory = options.RemoteFactory ?? (connector => new Remote(connector));
+        var timeProvider = options.TimeProvider ?? TimeProvider.System;
 
         watchManager = new WatchManager(this, masterCts.Token);
 
         // Start async init
         subscriptions = new InternalSubscriptionManager(
             firstStatusMatching: WaitForStatus,
             resolveOfflineSyncStatus: ResolveOfflineSyncStatus,
-            subscriptionsCommand: async (payload) => await this.WriteTransaction(async tx =>
+            subscriptionsCommand: async (payload) => await WriteTransaction(async tx =>
                 {
                     await tx.Execute("SELECT powersync_control(?, ?) AS r", ["subscriptions", JsonConvert.SerializeObject(payload)]);
                 }));
@@ -226,9 +234,17 @@ public PowerSyncDatabase(PowerSyncDatabaseOptions options)
                         await WaitForReady();
                         await connector.UploadData(this);
                     },
+                    PostCheckpointRequest = connector is ICustomCheckpointRequestConnector checkpointConnector
+                        ? async (clientId, requestId) =>
+                        {
+                            await WaitForReady();
+                            return await checkpointConnector.PostCheckpointRequest(clientId, requestId);
+                        }
+                        : null,
                     RetryDelayMs = options.RetryDelayMs,
                     Subscriptions = options.Subscriptions,
                     CrudUploadThrottleMs = options.CrudUploadThrottleMs,
+                    TimeProvider = timeProvider,
                     Logger = Logger
                 });
 
@@ -451,16 +467,6 @@ public async Task Init()
         await WaitForReady();
     }
 
-    private RequiredAdditionalConnectionOptions resolveConnectionOptions(PowerSyncConnectionOptions? options)
-    {
-        var defaults = RequiredAdditionalConnectionOptions.DEFAULT_ADDITIONAL_CONNECTION_OPTIONS;
-        return new RequiredAdditionalConnectionOptions
-        {
-            RetryDelayMs = options?.RetryDelayMs ?? defaults.RetryDelayMs,
-            CrudUploadThrottleMs = options?.CrudUploadThrottleMs ?? defaults.CrudUploadThrottleMs,
-        };
-    }
-
     public async Task Connect(IPowerSyncBackendConnector connector, PowerSyncConnectionOptions? options = null)
     {
         await WaitForReady();
@@ -829,6 +835,9 @@ public class SQLWatchOptions
     /// </summary>
     public int? ThrottleMs { get; set; }
 
+    /// <summary>
+    /// If true, runs the query once when creating the watch. Defaults to false.
+    /// </summary>
     public bool TriggerImmediately { get; set; } = false;
 }
 

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

@@ -6,6 +6,7 @@ namespace PowerSync.Common.Client.Sync.Bucket;
 
 using Newtonsoft.Json;
 
+using PowerSync.Common.DB;
 using PowerSync.Common.DB.Crud;
 using PowerSync.Common.Utils;
 
@@ -19,8 +20,6 @@ public static class PowerSyncControlCommand
     public const string NOTIFY_CRUD_UPLOAD_COMPLETED = "completed_upload";
     public const string UPDATE_SUBSCRIPTIONS = "update_subscriptions";
 
-    public const string TARGET_CHECKPOINT_REQUEST_ID = "target_checkpoint_request_id";
-
     /// <summary>
     /// An `established` or `end` event for response streams.
     /// </summary>
@@ -136,9 +135,14 @@ public interface IBucketStorageAdapter : ICloseable
     Task<CrudBatch?> GetCrudBatch(int limit = 100);
 
     Task<bool> UpdateLocalTarget(Func<Task<string>> callback);
-
     Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null);
 
+    // TODO Return int64 from this in future release
+    /// <summary>
+    /// Reads or updates the local checkpoint request ID counter.
+    /// </summary>
+    Task<string?> ReadOrUpdateCheckpoint(string variant, string? update = null);
+
     /// <summary>
     /// Get a unique client ID.
     /// </summary>
@@ -149,3 +153,47 @@ public interface IBucketStorageAdapter : ICloseable
     /// </summary>
     Task<string> Control(string op, object? payload);
 }
+
+/// <summary>
+/// Provides type-safe wrappers for <see cref="IBucketStorageAdapter.ReadOrUpdateCheckpoint" />.
+/// <para />
+/// Default Interface Implementations would be preferred here, but <c>netstandard2.0</c> doesn't
+/// support them.
+/// </summary>
+public static class BucketStorageAdapterExtensions
+{
+    /// <summary>
+    /// Increments and returns the local checkpoint counter.
+    /// </summary>
+    public static Task<string> NextCheckpointRequestId(this IBucketStorageAdapter adapter)
+        => adapter.ReadOrUpdateCheckpoint("next")!;
+
+    /// <summary>
+    /// Returns the highest checkpoint request ID that has been requested on this device.
+    /// </summary>
+    public static Task<string?> CurrentCheckpointRequestId(this IBucketStorageAdapter adapter)
+        => adapter.ReadOrUpdateCheckpoint("current");
+
+    /// <summary>
+    /// Seeds the local checkpoint request ID counter using a response from the server.
+    ///
+    /// Seeding the local counter achieves two goals:
+    /// <list type="number">
+    ///     <item>
+    ///         <description>
+    ///             The service is allowed to forget our checkpoint counter, so we remind
+    ///             it whenever we connect.
+    ///         </description>
+    ///     </item>
+    ///     <item>
+    ///         <description>
+    ///             Checkpoint requests are scoped per user-and-device combo, but the
+    ///             local ID counter is scoped per-device. Seeding ensures we generate
+    ///             correctly incrementing IDs after switching user accounts.
+    ///         </description>
+    ///     </item>
+    /// </list>
+    /// </summary>
+    public static Task<string> SeedCheckpointRequestId(this IBucketStorageAdapter adapter, string serviceResponse)
+        => adapter.ReadOrUpdateCheckpoint("seed", serviceResponse)!;
+}

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

@@ -10,7 +10,6 @@ namespace PowerSync.Common.Client.Sync.Bucket;
 
 using Newtonsoft.Json;
 
-using PowerSync.Common.Client.Sync.Stream;
 using PowerSync.Common.DB;
 using PowerSync.Common.DB.Crud;
 
@@ -69,18 +68,26 @@ public async Task<string> GetClientId()
     }
 
     /// <summary>
-    /// Reads the stored target checkpoint request id, or updates it when the update parameter is set.
+    /// Reads or updates the stored checkpoint request id.
     /// </summary>
-    /// <returns>The previous checkpoint request.</returns>
-    private static Task<string?> TargetCheckpointRequestId(ILockContext tx, string? update = null)
+    public Task<string?> ReadOrUpdateCheckpoint(string variant, string? update = null)
+        => db.WriteTransaction(tx => ReadOrUpdateCheckpoint(tx, variant, update));
+
+    /// <summary>
+    /// Reads or updates the stored checkpoint request id using the given transaction.
+    /// </summary>
+    public static Task<string?> ReadOrUpdateCheckpoint(ITransaction tx, string variant, string? payload = null)
     {
-        // TODO Note that we are only casting in Dart/JS because this returns a 64-bit integer we can't natively represent there.
-        // Turning MAX_OP_ID into a 64-bit integer here and comparing ints would be better.
+        // TODO Return 64-bit integer in later release.
         return tx.Get<string?>(
             "SELECT CAST(powersync_control(?, ?) AS TEXT) AS r",
-            [PowerSyncControlCommand.TARGET_CHECKPOINT_REQUEST_ID, update]);
+            [$"{variant}_checkpoint_request_id", payload]);
     }
 
+    // This is called within existing transactions, therefore accept an ITransaction instead of creating a new one
+    private static Task<string?> TargetCheckpointRequestId(ITransaction tx, string? update = null)
+        => ReadOrUpdateCheckpoint(tx, "target", update);
+
     private record ResultResult(object result);
 
     public class ResultDetail
@@ -154,6 +161,7 @@ public async Task<bool> UpdateLocalTarget(Func<Task<string>> callback)
             return true;
         });
     }
+
     public Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null)
     {
         return db.WriteTransaction(async tx =>

PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs

@@ -0,0 +1,23 @@
+namespace PowerSync.Common.Client.Sync;
+
+/// <summary>An exception related to checkpoint requests.</summary>
+public class CheckpointRequestException : Exception
+{
+    /// <summary>Initializes a new instance of the <see cref="CheckpointRequestException" /> class.</summary>
+    public CheckpointRequestException() : base() { }
+
+    /// <summary>Initializes a new instance of the <see cref="CheckpointRequestException" /> class with a specified error message.</summary>
+    public CheckpointRequestException(string message) : base(message) { }
+
+    /// <summary>Initializes a new instance of the <see cref="CheckpointRequestException" /> class with a specified error message and a reference to the inner exception that is the cause of this exception.</summary>
+    public CheckpointRequestException(string message, Exception innerException) : base(message, innerException) { }
+
+    /// <summary>The connected PowerSync Service does not support checkpoint requests.</summary>
+    public static readonly string InstanceNotSupported = "The PowerSync service does not support checkpoint requests. Update to PowerSync service version 1.24.0 or later to use this API.";
+
+    /// <summary>The sync client is disconnected.</summary>
+    public static readonly string Disconnected = "Cannot request checkpoints, sync client is disconnected";
+
+    /// <summary>Checkpoint requests are disabled; legacy write checkpoints are enabled.</summary>
+    public static readonly string Disabled = "Connected with legacy checkpoint mode, cannot request checkpoints";
+}