diff --git a/.github/workflows/dev-packages.yml b/.github/workflows/dev-packages.yml index 424d0e8a..e63e2fe6 100644 --- a/.github/workflows/dev-packages.yml +++ b/.github/workflows/dev-packages.yml @@ -22,7 +22,7 @@ jobs: - name: Setup .NET SDK uses: actions/setup-dotnet@v4 with: - dotnet-version: "8.0.x" + dotnet-version: "10.0.x" - name: Install MAUI Workloads run: dotnet workload restore diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 8847ae65..71c0874a 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -26,7 +26,7 @@ jobs: - name: Setup .NET SDK uses: actions/setup-dotnet@v4 with: - dotnet-version: "8.0.x" + dotnet-version: "10.0.x" - name: Install MAUI Workloads run: dotnet workload restore diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 15bc561c..110b65f1 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -18,7 +18,7 @@ jobs: - name: Setup .NET SDK uses: actions/setup-dotnet@v5 with: - dotnet-version: '8.0' + dotnet-version: '10.0' - name: Install MAUI Workloads run: dotnet workload restore @@ -30,4 +30,4 @@ jobs: run: dotnet restore - name: Run tests - run: dotnet test -v n --framework net8.0 \ No newline at end of file + run: dotnet test -v n --framework net10.0 diff --git a/Directory.build.props b/Directory.build.props index fe14bb0e..fed4b7a7 100644 --- a/Directory.build.props +++ b/Directory.build.props @@ -2,9 +2,5 @@ $(MSBuildWarningsAsMessages);NETSDK1202 - true - - - diff --git a/IsExternalInit.cs b/IsExternalInit.cs deleted file mode 100644 index 7d866a84..00000000 --- a/IsExternalInit.cs +++ /dev/null @@ -1,7 +0,0 @@ -using System.ComponentModel; - -namespace System.Runtime.CompilerServices -{ - [EditorBrowsable(EditorBrowsableState.Never)] - internal class IsExternalInit { } -} diff --git a/PowerSync/PowerSync.Common/CHANGELOG.md b/PowerSync/PowerSync.Common/CHANGELOG.md index 2d2e3edf..0e1b628c 100644 --- a/PowerSync/PowerSync.Common/CHANGELOG.md +++ b/PowerSync/PowerSync.Common/CHANGELOG.md @@ -10,6 +10,14 @@ - Accidental release (mirror of 0.1.2). Use 1.0.1 instead. +## 0.1.5 + +- Update the PowerSync SQLite core extension to 0.5.3. + - Fix a transaction affecting both insert-only and regular tables being recorded as two transaction ids. + - Improve error messages for errors originating from the core extension. +- Add `PowerSyncConnectionOptions.CheckpointMode`. When set to `CheckpointMode.Requests()`, the connection manager uses a new protocol for checkpoints after crud uploads with better support for switching users. +- Add `PowerSyncDatabase.RequestCheckpoint()`, which can be used to request sync updates explicitly and wait for those to complete. + ## 0.1.4 - Update the PowerSync SQLite core extension to 0.5.2. diff --git a/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs b/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs index a8fdb65e..cb286e6f 100644 --- a/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs +++ b/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs @@ -25,3 +25,24 @@ public interface IPowerSyncBackendConnector /// Task UploadData(IPowerSyncDatabase database); } + +/// +/// An capable of requesting checkpoints. +/// +/// Extend this class instead of 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 asynchronous backend uploads. +/// +/// To use this connector, using is required. Note that +/// this requires PowerSync service version 1.24.0 or later. +/// +public interface ICustomCheckpointRequestConnector : IPowerSyncBackendConnector +{ + /// + /// Posts a client-generated checkpoint request to the backend and returns the effective checkpoint request state. + /// + Task PostCheckpointRequest(string clientId, long requestId, CancellationToken token); +} diff --git a/PowerSync/PowerSync.Common/Client/ConnectionManager.cs b/PowerSync/PowerSync.Common/Client/ConnectionManager.cs index 50b6ce7d..4b86f952 100644 --- a/PowerSync/PowerSync.Common/Client/ConnectionManager.cs +++ b/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). diff --git a/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs b/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs index b8cc8e87..49383d3a 100644 --- a/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs +++ b/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs @@ -11,6 +11,7 @@ namespace PowerSync.Common.Client; using Nito.AsyncEx; using PowerSync.Common.Client.Connection; +using PowerSync.Common.Client.Sync; using PowerSync.Common.Client.Sync.Bucket; using PowerSync.Common.Client.Sync.Stream; using PowerSync.Common.DB; @@ -50,6 +51,12 @@ public class PowerSyncDatabaseOptions() : BasePowerSyncDatabaseOptions() /// If not provided, a default Remote will be created. /// public Func? RemoteFactory { get; set; } + + /// + /// Used to calculate delays for the sync client (retry delays, upload throttling). + /// Used for testing to avoid waiting out some long delays (retry delay is minimum 10 seconds). + /// + internal TimeProvider? TimeProvider { get; set; } } public class PowerSyncDBEvents : EventManager @@ -200,6 +207,7 @@ public PowerSyncDatabase(PowerSyncDatabaseOptions options) SdkVersion = ""; remoteFactory = options.RemoteFactory ?? (connector => new Remote(connector)); + var timeProvider = options.TimeProvider ?? TimeProvider.System; watchManager = new WatchManager(this, masterCts.Token); @@ -207,7 +215,7 @@ public PowerSyncDatabase(PowerSyncDatabaseOptions options) 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, token) => + { + await WaitForReady(); + return await checkpointConnector.PostCheckpointRequest(clientId, requestId, token); + } + : null, RetryDelayMs = options.RetryDelayMs, Subscriptions = options.Subscriptions, CrudUploadThrottleMs = options.CrudUploadThrottleMs, + TimeProvider = timeProvider, Logger = Logger }); @@ -319,6 +335,7 @@ public async Task WaitForStatus(Func predicate, CancellationTo } var tcs = new TaskCompletionSource(); + var canceledRegistration = cts.Token.Register(() => tcs.TrySetCanceled(cts.Token)); _ = Task.Run(async () => { @@ -334,9 +351,22 @@ public async Task WaitForStatus(Func predicate, CancellationTo } } catch (OperationCanceledException) { } + catch (Exception ex) + { + tcs.TrySetException(ex); + cts.Cancel(); + } }); - await tcs.Task; + try + { + await tcs.Task; + } + finally + { + canceledRegistration.Dispose(); + cts.Cancel(); + } } protected async Task Initialize(PowerSyncDatabaseOptions options) @@ -451,16 +481,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(); @@ -506,6 +526,41 @@ await Database.WriteTransaction(async tx => Events.Emit(new PowerSyncDBEvents.StatusChangedEvent(CurrentStatus)); } + /// + /// Requests a checkpoint from the PowerSync service. + /// + /// The returned request can be awaited using + /// to confirm that the local database has applied server-side changes up to + /// the checkpoint. This method requires an active or connecting sync client + /// connected with and PowerSync service version + /// 1.24.0 or later. + /// + /// + /// Thrown when requesting the checkpoint has failed, for example when the + /// database is disconnected. + /// + public async Task RequestCheckpoint(CancellationToken ct = default) + { + await WaitForReady(); + + // Important: `LockAsync(CancellationToken)` will still take the lock even if + // cancellation is requested, so long as the token is canceled before LockAsync + // is called and the lock is not in use; i.e. an already-canceled token will only + // prevent _waiting_ for the lock, not getting the lock. Therefore, we need to + // check for cancellation ourselves before entering the using block. + ct.ThrowIfCancellationRequested(); + using (await runExclusive.LockAsync(ct)) + { + var sync = SyncStreamImplementation; + if (sync == null) + { + throw new CheckpointRequestException(CheckpointRequestException.Disconnected); + } + + return await sync.RequestCheckpoint(this, ct); + } + } + /// /// Create a sync stream to query its status or to subscribe to it. /// @@ -829,6 +884,9 @@ public class SQLWatchOptions /// public int? ThrottleMs { get; set; } + /// + /// If true, runs the query once when creating the watch. Defaults to false. + /// public bool TriggerImmediately { get; set; } = false; } diff --git a/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs index 5a8e1c45..23a158e6 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs @@ -19,8 +19,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"; - /// /// An `established` or `end` event for response streams. /// @@ -136,9 +134,45 @@ public interface IBucketStorageAdapter : ICloseable Task GetCrudBatch(int limit = 100); Task UpdateLocalTarget(Func> callback); - Task HandleCrudCheckpoint(long lastClientId, long? writeCheckpoint = null); + /// + /// Reads or updates the local checkpoint request ID counter. + /// + Task ReadOrUpdateCheckpoint(string variant, long? update = null); + + /// + /// Increments and returns the local checkpoint counter. + /// + public async Task NextCheckpointRequestId() => (long)await ReadOrUpdateCheckpoint("next"); + + /// + /// Returns the highest checkpoint request ID that has been requested on this device. + /// + public Task CurrentCheckpointRequestId() => ReadOrUpdateCheckpoint("current"); + + /// + /// Seeds the local checkpoint request ID counter using a response from the server. + /// + /// Seeding the local counter achieves two goals: + /// + /// + /// + /// The service is allowed to forget our checkpoint counter, so we remind + /// it whenever we connect. + /// + /// + /// + /// + /// 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. + /// + /// + /// + /// + public async Task SeedCheckpointRequestId(long serviceResponse) => (long)await ReadOrUpdateCheckpoint("seed", serviceResponse); + /// /// Get a unique client ID. /// @@ -148,4 +182,6 @@ public interface IBucketStorageAdapter : ICloseable /// Invokes the `powersync_control` function for the sync client. /// Task Control(string op, object? payload); + + } diff --git a/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs b/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs index 88235e7f..21dbd193 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs @@ -10,14 +10,11 @@ namespace PowerSync.Common.Client.Sync.Bucket; using Newtonsoft.Json; -using PowerSync.Common.Client.Sync.Stream; using PowerSync.Common.DB; using PowerSync.Common.DB.Crud; public class SqliteBucketStorage : IBucketStorageAdapter { - public const long MAX_OP_ID = 9223372036854775807; - public BucketStorageEvents Events { get; } = new(); private readonly IDBAdapter db; @@ -69,16 +66,25 @@ public async Task GetClientId() } /// - /// Reads the stored target checkpoint request id, or updates it when the update parameter is set. + /// Reads or updates the stored checkpoint request id. + /// + public Task ReadOrUpdateCheckpoint(string variant, long? update = null) + => db.WriteTransaction(tx => ReadOrUpdateCheckpoint(tx, variant, update)); + + /// + /// Reads or updates the stored checkpoint request id using the given transaction. /// - /// The previous checkpoint request. - private static Task TargetCheckpointRequestId(ILockContext tx, long? update = null) + public static Task ReadOrUpdateCheckpoint(ITransaction tx, string variant, long? payload = null) { return tx.Get( "SELECT powersync_control(?, ?) 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 TargetCheckpointRequestId(ITransaction tx, long? update = null) + => ReadOrUpdateCheckpoint(tx, "target", update); + private record ResultResult(object result); public class ResultDetail @@ -97,7 +103,7 @@ public async Task UpdateLocalTarget(Func> callback) var seqBeforeResult = await db.ReadTransaction(async tx => { var currentTarget = await TargetCheckpointRequestId(tx); - if (currentTarget != MAX_OP_ID) + if (currentTarget != long.MaxValue) { // Nothing to update return (long?)null; @@ -152,6 +158,7 @@ public async Task UpdateLocalTarget(Func> callback) return true; }); } + public Task HandleCrudCheckpoint(long lastClientId, long? writeCheckpoint = null) { return db.WriteTransaction(async tx => @@ -163,7 +170,7 @@ public Task HandleCrudCheckpoint(long lastClientId, long? writeCheckpoint = null await TargetCheckpointRequestId( tx, - writeCheckpoint is not null && !crudRemaining ? writeCheckpoint : MAX_OP_ID); + writeCheckpoint is not null && !crudRemaining ? writeCheckpoint : long.MaxValue); }); } diff --git a/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs b/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs new file mode 100644 index 00000000..25433594 --- /dev/null +++ b/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs @@ -0,0 +1,110 @@ +using PowerSync.Common.Client.Sync.Stream; + +namespace PowerSync.Common.Client.Sync; + +/// +/// A checkpoint request created by . +/// +/// Use this value to wait until the local database has applied server-side changes up to the +/// requested checkpoint. This is useful for explicit refresh flows where the caller wants +/// confirmation that the local view has caught up to the service. +/// +/// Checkpoint requests are backed by request ids tracked in the local database, so they are reusable +/// across disconnect and reconnect cycles. A wait interrupted by a disconnect throws an error, but +/// the same request can be awaited again once a new connection is established. +/// +/// Requests do not survive , instances created +/// before a clear should be discarded and requested again. +/// +public class CheckpointRequest +{ + private readonly long _requestId; + private readonly PowerSyncDatabase _db; + + internal CheckpointRequest(long requestId, PowerSyncDatabase db) + { + _requestId = requestId; + _db = db; + } + + ///Whether this checkpoint request has synced before. + public bool HasSynced { get => _db.SyncStreamImplementation?.IsCheckpointRequestApplied(_requestId) ?? false; } + + /// + /// Waits until this checkpoint has been synced locally. + /// + /// This method fails on sync errors: If a download or upload error occurs before this checkpoint + /// request has synced, that error is rethrown here. This makes it easier to observe sync errors + /// when relying on checkpoints. Once sync has recovered, it is valid to call this method again + /// to await the checkpoint. + /// + /// + /// + /// Thrown if the cancellation token is canceled. Importantly, this is not thrown if the checkpoint + /// has already finished syncing, as there is no work to be canceled. + /// + public Task WaitForSync(CancellationToken ct = default) + { + if (HasSynced) return Task.CompletedTask; + + ct.ThrowIfCancellationRequested(); + + var sync = _db.SyncStreamImplementation; + + if (sync is null) + { + throw new CheckpointRequestException(CheckpointRequestException.Disconnected); + } + + if (sync.ConnectionOptions?.CheckpointMode == CheckpointMode.Legacy) + { + throw new CheckpointRequestException(CheckpointRequestException.Disabled); + } + + return _db.WaitForStatus(status => + { + if (HasSynced) + { + return true; + } + + Exception? anyError = status.DataFlowStatus.DownloadError ?? status.DataFlowStatus.UploadError; + if (anyError is not null) + { + throw new CheckpointRequestException(CheckpointRequestException.StatusError, anyError); + } + + if (!status.Connected && !status.Connecting) + { + throw new CheckpointRequestException(CheckpointRequestException.Disconnected); + } + + return false; + }, ct); + } +} + +/// An exception related to checkpoint requests. +public class CheckpointRequestException : Exception +{ + /// Initializes a new instance of the class. + public CheckpointRequestException() : base() { } + + /// Initializes a new instance of the class with a specified error message. + public CheckpointRequestException(string message) : base(message) { } + + /// Initializes a new instance of the class with a specified error message and a reference to the inner exception that is the cause of this exception. + public CheckpointRequestException(string message, Exception innerException) : base(message, innerException) { } + + /// The connected PowerSync Service does not support checkpoint requests. + 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."; + + /// The sync client is disconnected. + public static readonly string Disconnected = "Cannot request checkpoints, sync client is disconnected"; + + /// Checkpoint requests are disabled; legacy write checkpoints are enabled. + public static readonly string Disabled = "Connected with legacy checkpoint mode, cannot request checkpoints"; + + /// + public static readonly string StatusError = "Error on sync status before checkpoint was applied"; +} diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/CheckpointState.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/CheckpointState.cs new file mode 100644 index 00000000..4147081d --- /dev/null +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/CheckpointState.cs @@ -0,0 +1,170 @@ +namespace PowerSync.Common.Client.Sync.Stream; + +using System.Runtime.ExceptionServices; +using System.Threading.Channels; + +using PowerSync.Common.Utils; + +/// +/// Tracks whether the active download iteration has reconciled checkpoint request state with the +/// PowerSync service, gating checkpoint requests until it has. +/// +internal sealed class CheckpointStateSignals +{ + private CheckpointState _state = new CheckpointState.Pending(); + + private readonly BroadcastChannel _stateBroadcaster = new(); + private Channel _checkpointWaiterNotifier = CreateNotifier(); + + private readonly object _lock = new(); + + /// + /// Marks the current download iteration as ended, blocking new checkpoint requests until the + /// seed performed by the next iteration completes. + /// + public void DownloadIterationEnded() + { + lock (_lock) + { + // Waiters arriving after this should be able to resume the next download iteration. + _checkpointWaiterNotifier = CreateNotifier(); + UpdateState(new CheckpointState.Pending()); + } + } + + /// + /// Marks the sync client as disconnected, failing all outstanding checkpoint requests and + /// preventing new ones. + /// + public void Disconnected() + { + lock (_lock) + { + UpdateState(new CheckpointState.Disconnected()); + } + } + + /// + /// Runs , publishing its outcome to callers of + /// . Cancellation leaves the state pending, since a + /// later iteration will seed it again. + /// + public async Task MarkCheckpointsReady(Func seed) + { + try + { + await seed(); + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception ex) + { + lock (_lock) + { + UpdateState(new CheckpointState.Failed(ex)); + } + throw; + } + + lock (_lock) + { + UpdateState(new CheckpointState.Ready()); + } + } + + /// + /// Waits for a caller wanting to request a checkpoint. + /// + /// That caller is blocked until the seed run started by a download iteration completes, so this + /// is used to wake up the download loop while it is paused between iterations. + /// + public async Task WaitForCheckpointWaiter(CancellationToken signal) + { + ChannelReader reader; + lock (_lock) + { + reader = _checkpointWaiterNotifier.Reader; + } + + await reader.ReadAsync(signal); + } + + /// + /// Waits until a download iteration is active and has seeded the checkpoint state, meaning that + /// checkpoint request ids can safely be allocated. + /// + public async Task WaitForCheckpointRequestsReady(CancellationToken signal, bool wakeDownloadLoop = true) + { + var reader = _stateBroadcaster.Subscribe(out var subscriberId); + try + { + // Only the first check may wake the download loop. A waiter that is already parked when + // an iteration ends must not resume the next one, otherwise a download loop that keeps + // failing would retry without ever waiting out its retry delay. + var wake = wakeDownloadLoop; + + while (!CheckpointRequestsReady(wake)) + { + wake = false; + await reader.ReadAsync(signal); + } + } + finally + { + _stateBroadcaster.Unsubscribe(subscriberId); + } + } + + /// + /// Returns true if checkpoint requests are ready and false if we need to keep waiting. + /// + private bool CheckpointRequestsReady(bool wakeDownloadLoop) + { + lock (_lock) + { + switch (_state) + { + case CheckpointState.Ready: + return true; + case CheckpointState.Disconnected: + throw new CheckpointRequestException(CheckpointRequestException.Disconnected); + case CheckpointState.Failed failed: + ExceptionDispatchInfo.Capture(failed.Exception).Throw(); + return true; + case CheckpointState.Pending: + if (wakeDownloadLoop) + { + _checkpointWaiterNotifier.Writer.TryWrite(true); + } + return false; + default: + throw new InvalidOperationException($"Invalid CheckpointState: {_state}"); + } + } + } + + private void UpdateState(CheckpointState next) + { + _state = next; + _stateBroadcaster.Broadcast(true); + } + + /// Channel that always holds the latest item written. Used to notify listeners that an event has occured. + private static Channel CreateNotifier() => + Channel.CreateBounded(new BoundedChannelOptions(1) { FullMode = BoundedChannelFullMode.DropWrite }); +} + +internal abstract record CheckpointState +{ + private CheckpointState() { } + + public sealed record Pending : CheckpointState; + + public sealed record Disconnected : CheckpointState; + + public sealed record Ready : CheckpointState; + + public sealed record Failed(Exception Exception) : CheckpointState; +} diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs index 2f94ae5d..efc5adf7 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs @@ -2,6 +2,7 @@ using Newtonsoft.Json.Linq; using PowerSync.Common.DB.Crud; +using PowerSync.Common.Utils.Converters; namespace PowerSync.Common.Client.Sync.Stream; @@ -21,7 +22,7 @@ public static Instruction[] ParseInstructions(string rawResponse) instructions.Add(ParseInstruction(item)); } - return instructions.ToArray(); + return [.. instructions]; } public static Instruction ParseInstruction(JObject json) @@ -61,6 +62,9 @@ public class EstablishSyncStream : Instruction { [JsonProperty("request")] public StreamingSyncRequest Request { get; set; } = null!; + + [JsonProperty("checkpoint_request", NullValueHandling = NullValueHandling.Ignore)] + public CheckpointRequestPayload? CheckpointRequest { get; set; } = null!; } public class UpdateSyncStatus : NonInterruptingInstruction @@ -129,6 +133,9 @@ public class CoreSyncStatus [JsonProperty("streams")] public List Streams { get; set; } = []; + + [JsonProperty("internal_last_applied_checkpoint_request_id")] + public long? LastAppliedCheckpointRequestId { get; set; } } public class SyncPriorityStatus diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/Remote.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/Remote.cs index c253d34d..4029bcb9 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/Remote.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/Remote.cs @@ -102,12 +102,12 @@ static string GetUserAgent() return $"powersync-dotnet/{version}"; } - public virtual async Task Get(string path, Dictionary? headers = null) + // TODO: Potentially use an abstract base class (similar to JS) instead of making arbitrary virtual + public virtual async Task FetchJson(string path, HttpMethod? method = null, object? data = null, Dictionary? headers = null, CancellationToken ct = default) { - var request = await BuildRequest(HttpMethod.Get, path, data: null, additionalHeaders: headers); + var request = await BuildRequest(method ?? HttpMethod.Get, path, data, headers); - using var client = new HttpClient(); - var response = await client.SendAsync(request); + var response = await httpClient.SendAsync(request, ct); if (response.StatusCode == System.Net.HttpStatusCode.Unauthorized) { @@ -129,8 +129,8 @@ public virtual async Task Get(string path, Dictionary? hea /// public virtual async Task PostStreamRaw(SyncStreamOptions options) { - var requestMessage = await BuildRequest(HttpMethod.Post, options.Path, options.Data, options.Headers); - var response = await httpClient.SendAsync(requestMessage, HttpCompletionOption.ResponseHeadersRead, options.CancellationToken); + var request = await BuildRequest(HttpMethod.Post, options.Path, options.Data, options.Headers); + var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, options.CancellationToken); if (response.Content == null) { diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs index 350c99b0..3234950c 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs @@ -3,6 +3,7 @@ namespace PowerSync.Common.Client.Sync.Stream; using System.Net.Sockets; using System.Text; using System.Threading.Tasks; +using System.Threading.Channels; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; @@ -16,15 +17,12 @@ namespace PowerSync.Common.Client.Sync.Stream; public class AdditionalConnectionOptions(int? retryDelayMs = null, int? crudUploadThrottleMs = null) { /// - /// Delay for retrying sync streaming operations - /// from the PowerSync backend after an error occurs. + /// Delay for retrying sync streaming operations from the PowerSync backend after an error occurs. /// public int? RetryDelayMs { get; set; } = retryDelayMs; /// - /// Backend Connector CRUD operations are throttled - /// to occur at most every `CrudUploadThrottleMs` - /// milliseconds. + /// Backend Connector CRUD operations are throttled to occur at most every `CrudUploadThrottleMs` milliseconds. /// public int? CrudUploadThrottleMs { get; set; } = crudUploadThrottleMs; } @@ -43,7 +41,6 @@ public class RequiredAdditionalConnectionOptions : AdditionalConnectionOptions public new int CrudUploadThrottleMs { get; set; } public SubscribedStream[] Subscriptions { get; init; } = null!; - } public class StreamingSyncImplementationOptions : AdditionalConnectionOptions @@ -54,12 +51,24 @@ public class StreamingSyncImplementationOptions : AdditionalConnectionOptions public Func UploadCrud { get; init; } = null!; + /// + /// Posts a checkpoint request with the connector. Null when the connector doesn't support that, + /// in which case the request is posted to the PowerSync service directly. + /// + public Func>? PostCheckpointRequest { get; init; } + public Remote Remote { get; init; } = null!; public ILogger? Logger { get; init; } + + /// + /// Source of the delays in the sync loops. Tests substitute a fake clock so they don't have to + /// wait out real retry delays. + /// + internal TimeProvider TimeProvider { get; init; } = TimeProvider.System; } -public class BaseConnectionOptions(Dictionary? parameters = null, Dictionary? appMetadata = null, bool? includeDefaultStreams = true) +public class BaseConnectionOptions(Dictionary? parameters = null, Dictionary? appMetadata = null, bool? includeDefaultStreams = true, CheckpointMode? checkpointMode = null) { /// /// A set of metadata to be included in service logs. @@ -77,11 +86,17 @@ public class BaseConnectionOptions(Dictionary? parameters = null /// This defaults to `true`. /// public bool? IncludeDefaultStreams { get; set; } = includeDefaultStreams; + + /// + /// The mode used to request checkpoint requests from the PowerSync service. + /// + /// Defaults to , but will default to in a future release. + /// + public CheckpointMode CheckpointMode { get; set; } = checkpointMode ?? CheckpointMode.Legacy; } public class RequiredPowerSyncConnectionOptions : BaseConnectionOptions { - public new Dictionary AppMetadata { get; set; } = new(); public new Dictionary Params { get; set; } = new(); @@ -124,8 +139,9 @@ public class PowerSyncConnectionOptions( int? retryDelayMs = null, int? crudUploadThrottleMs = null, Dictionary? appMetadata = null, - bool? includeDefaultStreams = true -) : BaseConnectionOptions(@params, appMetadata, includeDefaultStreams) + bool? includeDefaultStreams = true, + CheckpointMode? checkpointMode = null +) : BaseConnectionOptions(@params, appMetadata, includeDefaultStreams, checkpointMode) { /// /// Delay for retrying sync streaming operations from the PowerSync backend after an error occurs. @@ -145,16 +161,16 @@ public class SubscribedStream [JsonProperty("params")] public Dictionary? Params { get; set; } - } public class StreamingSyncImplementation : ICloseable { - public static RequiredPowerSyncConnectionOptions DEFAULT_STREAM_CONNECTION_OPTIONS = new() + public static readonly RequiredPowerSyncConnectionOptions DEFAULT_STREAM_CONNECTION_OPTIONS = new() { AppMetadata = [], Params = [], - IncludeDefaultStreams = true + IncludeDefaultStreams = true, + CheckpointMode = CheckpointMode.Legacy, }; public StreamingSyncImplementationEvents Events { get; } = new(); @@ -162,24 +178,38 @@ public class StreamingSyncImplementation : ICloseable public static readonly int DEFAULT_CRUD_UPLOAD_THROTTLE_MS = 1000; public static readonly int DEFAULT_RETRY_DELAY_MS = 5000; - protected StreamingSyncImplementationOptions Options { get; } + protected internal StreamingSyncImplementationOptions Options { get; } + + protected internal PowerSyncConnectionOptions? ConnectionOptions { get; private set; } protected CancellationTokenSource? CancellationTokenSource { get; set; } private Task? streamingSyncTask; - public Action TriggerCrudUpload { get; } private CancellationTokenSource? crudUpdateCts; private Task? crudUpdateTask; + private readonly CheckpointStateSignals checkpointState = new(); + + /// + /// The highest checkpoint request id the core extension has reported as applied, if any. + /// + internal long? LastAppliedCheckpointRequestId; + private readonly ILogger logger; private SubscribedStream[] activeStreams; - private bool isUploadingCrud; - private Task? crudUploadTask; - private Action? notifyCompletedUploads; private Action? handleActiveStreamsChange; + /// Signals that there may be local writes to upload. + private readonly Channel crudUploadRequested = CreateNotifier(); + + /// + /// Signals that a CRUD upload pass finished, so the core extension can re-check whether it still + /// has to hold a checkpoint back for pending local writes. + /// + private readonly Channel crudUploadCompleted = CreateNotifier(); + private readonly StreamingSyncLocks locks; public StreamingSyncImplementation(StreamingSyncImplementationOptions options) @@ -200,25 +230,8 @@ public StreamingSyncImplementation(StreamingSyncImplementationOptions options) locks = new StreamingSyncLocks(); logger = options.Logger ?? NullLogger.Instance; - isUploadingCrud = false; CancellationTokenSource = null; - - TriggerCrudUpload = () => - { - if (!SyncStatus.Connected || isUploadingCrud) - { - return; - } - - isUploadingCrud = true; - crudUploadTask = Task.Run(async () => - { - await InternalUploadAllCrud(); - notifyCompletedUploads?.Invoke(); - isUploadingCrud = false; - }); - }; } /// @@ -226,7 +239,6 @@ public StreamingSyncImplementation(StreamingSyncImplementationOptions options) /// public bool IsConnected => SyncStatus.Connected; - /// /// The timestamp of the last successful sync. /// @@ -293,6 +305,7 @@ public async Task Disconnect() await streamingSyncTask; } } + catch (OperationCanceledException) { } catch (Exception ex) { // The operation might have failed, all we care about is if it has completed @@ -301,21 +314,80 @@ public async Task Disconnect() streamingSyncTask = null; CancellationTokenSource = null; - // Do the same for any pending CRUD uploads - if (crudUploadTask != null) + UpdateSyncStatus(new SyncStatusOptions { Connected = false, Connecting = false }); + } + + /// + /// Requests a CRUD upload, without waiting for it to complete. + /// + public void TriggerCrudUpload() + { + crudUploadRequested.Writer.TryWrite(true); + } + + internal async Task RequestCheckpoint(PowerSyncDatabase db, CancellationToken ct) + { + if (ConnectionOptions?.CheckpointMode == CheckpointMode.Legacy) { - try - { - await crudUploadTask; - } - catch (Exception ex) - { - logger.LogWarning("CRUD upload task failed during disconnect: {Message}", ex.Message); - } - crudUploadTask = null; + throw new CheckpointRequestException(CheckpointRequestException.Disabled); } - UpdateSyncStatus(new SyncStatusOptions { Connected = false, Connecting = false }); + var requestId = await RequestNextCheckpointFromService(ct); + return new CheckpointRequest(requestId, db); + } + + /// + /// Allocates the next checkpoint request id and posts it, waiting for the active download + /// iteration to have reconciled checkpoint state with the service first. + /// + private async Task RequestNextCheckpointFromService(CancellationToken signal) + { + await checkpointState.WaitForCheckpointRequestsReady(signal); + + var nextCheckpointRequestId = await Options.Adapter.NextCheckpointRequestId(); + var clientId = await Options.Adapter.GetClientId(); + return await RequestCheckpointFromService(signal, new CheckpointRequestPayload + { + ClientId = clientId, + CheckpointRequestId = nextCheckpointRequestId, + }); + } + + private async Task RequestCheckpointFromService(CancellationToken signal, CheckpointRequestPayload request) + { + // First, check if we can use a custom checkpoint request implementation. + if (Options.PostCheckpointRequest != null) + { + return await Options.PostCheckpointRequest(request.ClientId, request.CheckpointRequestId, signal); + } + + var status = await Options.Remote.FetchJson( + path: "/sync/checkpoint-request", + method: HttpMethod.Post, + data: request, + ct: signal + ); + return status.Data.CheckpointRequestId; + } + + /// + /// Asks the service for the checkpoint request state it has for this client, and hands it to the + /// core extension so that subsequent requests continue from a counter both parties agree on. + /// + private async Task SeedCheckpointRequestState(CancellationToken signal, CheckpointRequestPayload request) + { + var seed = await RequestCheckpointFromService(signal, request); + await Options.Adapter.SeedCheckpointRequestId(seed); + } + + private async Task GetLegacyWriteCheckpoint(CancellationToken signal) + { + var clientId = await Options.Adapter.GetClientId(); + var path = $"/write-checkpoint2.json?client_id={clientId}"; + var response = await Options.Remote.FetchJson(path, ct: signal); + + logger.LogDebug("Created write checkpoint: {checkpoint}", response.Data.WriteCheckpoint); + return response.Data.WriteCheckpoint; } protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectionOptions? options) @@ -326,6 +398,30 @@ protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectio signal = CancellationTokenSource.Token; } + var token = signal.Value; + var resolvedOptions = options ?? new PowerSyncConnectionOptions(); + ConnectionOptions = resolvedOptions; + + try + { + await Task.WhenAll( + DownloadLoop(token, resolvedOptions), + CrudUploadLoop(token, resolvedOptions), + RepostUnacknowledgedCheckpointRequests(token, resolvedOptions) + ); + } + finally + { + // These loops only complete when we want to disconnect. No further sync iteration can + // resume checkpoint requests, so fail any that are still pending. + checkpointState.Disconnected(); + } + } + + protected async Task DownloadLoop(CancellationToken signal, PowerSyncConnectionOptions options) + { + var retryDelayMs = options.RetryDelayMs ?? DEFAULT_RETRY_DELAY_MS; + crudUpdateCts = new CancellationTokenSource(); crudUpdateTask = Task.Run(async () => { @@ -338,12 +434,12 @@ protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectio // Create a new cancellation token source for nested operations. // This is needed to close any previous connections. var nestedCts = new CancellationTokenSource(); - signal.Value.Register(() => + signal.Register(() => { nestedCts.Cancel(); crudUpdateCts?.Cancel(); crudUpdateCts = null; - try { crudUpdateTask?.Wait(2000); } catch (Exception) { } + try { crudUpdateTask?.Wait(2000); } catch { } UpdateSyncStatus(new SyncStatusOptions { Connected = false, @@ -364,7 +460,7 @@ protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectio try { - if (signal.Value.IsCancellationRequested) + if (signal.IsCancellationRequested) { break; } @@ -409,9 +505,7 @@ protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectio } finally { - notifyCompletedUploads = null; - - if (!signal.Value.IsCancellationRequested) + if (!signal.IsCancellationRequested) { // Closing sync stream network requests before retry. nestedCts.Cancel(); @@ -426,7 +520,9 @@ protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectio Connecting = true }); - await DelayRetry(); + // Someone wanting to request a checkpoint needs a seeded iteration, so cut the + // delay short instead of making them wait for it. + await DelayRetry(signal, retryDelayMs, resumeOnCheckpointRequest: true); } } } @@ -439,6 +535,129 @@ protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectio }); } + /// + /// Uploads local writes for as long as the connection lasts: once on connect, and then whenever + /// signals that there may be more. + /// + protected async Task CrudUploadLoop(CancellationToken signal, PowerSyncConnectionOptions options) + { + var throttleMs = options.CrudUploadThrottleMs ?? DEFAULT_CRUD_UPLOAD_THROTTLE_MS; + + try + { + // This function bundles InternalUploadAllCrud and crudUploadCompleter.Write to + // prevent the write call from waiting on the retry delay. + async Task UploadAllCrudThenSignalCompletion() + { + try + { + await InternalUploadAllCrud(signal, options); + } + finally + { + crudUploadCompleted.Writer.TryWrite(true); + } + } + + while (!signal.IsCancellationRequested) + { + // Start the initial CRUD upload on connect. Then, keep polling until we're done. + // The throttle runs alongside the upload so that completing it isn't delayed by + // the remainder of the throttle. + await Task.WhenAll( + UploadAllCrudThenSignalCompletion(), + DelayRetry(signal, throttleMs) + ); + + await crudUploadRequested.Reader.ReadAsync(signal); + } + } + catch (OperationCanceledException) when (signal.IsCancellationRequested) { /* Disconnecting. */ } + catch (Exception ex) + { + logger.LogError("Error in CRUD upload loop: {message}", ex.Message); + } + } + + /// + /// Periodically re-posts the current checkpoint request while the service has not applied it yet. + /// + /// The service is allowed to forget checkpoint requests, and re-posting an id it has already seen + /// is a cheap no-op, so this doubles as a catch-all for requests lost to network failures. + /// + protected async Task RepostUnacknowledgedCheckpointRequests(CancellationToken signal, PowerSyncConnectionOptions options) + { + if (options.CheckpointMode is not CheckpointMode.Requests requests) + { + return; + } + + var retryDelayMs = requests.RetryDelayMs; + + while (!signal.IsCancellationRequested) + { + try + { + // Never wakes the download loop: this only re-posts what another caller requested. + await checkpointState.WaitForCheckpointRequestsReady(signal, wakeDownloadLoop: false); + + var requestId = await Options.Adapter.CurrentCheckpointRequestId(); + + // Give the request some time to sync. + await DelayRetry(signal, retryDelayMs); + + // If a new request was made, reset the timer. + if (requestId != await Options.Adapter.CurrentCheckpointRequestId()) + { + continue; + } + + // If the request was applied, we don't need to retry. + if (requestId == null || IsCheckpointRequestApplied(requestId.Value)) + { + continue; + } + + // Make sure we're online and ready before making the request. + await checkpointState.WaitForCheckpointRequestsReady(signal, wakeDownloadLoop: false); + + // It's safe if this request races with a new one, the service will reject it. + logger.LogDebug("Retry checkpoint request {requestId}", requestId.Value); + await RequestCheckpointFromService(signal, new CheckpointRequestPayload + { + ClientId = await Options.Adapter.GetClientId(), + CheckpointRequestId = requestId.Value, + }); + } + catch (OperationCanceledException) when (signal.IsCancellationRequested) + { + return; + } + catch (Exception ex) + { + logger.LogWarning("Error retrying checkpoint request: {message}", ex.Message); + + try + { + await DelayRetry(signal, retryDelayMs); + } + catch (OperationCanceledException) + { + return; + } + } + } + } + + /// + /// Whether the core extension has reported (or a later request) as + /// applied. + /// + internal bool IsCheckpointRequestApplied(long requestId) + { + return LastAppliedCheckpointRequestId is not null && LastAppliedCheckpointRequestId >= requestId; + } + protected record StreamingSyncIterationResult { public bool? LegacyRetry { get; init; } @@ -450,8 +669,13 @@ protected record EnqueuedCommand { public string Command { get; init; } = null!; public object? Payload { get; init; } - } + /// + /// Set instead of when work running alongside the iteration (seeding + /// checkpoint state) failed and the iteration should fail with it. + /// + public Exception? Error { get; init; } + } protected async Task StreamingSyncIteration(CancellationToken signal, PowerSyncConnectionOptions? options) { @@ -466,6 +690,7 @@ protected async Task StreamingSyncIteration(Cancel AppMetadata = options?.AppMetadata ?? DEFAULT_STREAM_CONNECTION_OPTIONS.AppMetadata, Params = options?.Params ?? DEFAULT_STREAM_CONNECTION_OPTIONS.Params, IncludeDefaultStreams = options?.IncludeDefaultStreams ?? DEFAULT_STREAM_CONNECTION_OPTIONS.IncludeDefaultStreams, + CheckpointMode = options?.CheckpointMode ?? DEFAULT_STREAM_CONNECTION_OPTIONS.CheckpointMode, }; return await RustStreamingSyncIteration(signal, resolvedOptions); @@ -482,6 +707,9 @@ protected async Task RustStreamingSyncIteration(Ca // A failure opening or reading the stream, surfaced from the control loop so it retries. Exception? streamError = null; + // Reconciling checkpoint request state runs alongside line processing rather than blocking it. + Task? seedingCheckpointState = null; + var nestedCts = new CancellationTokenSource(); signal?.Register(() => { nestedCts.Cancel(); }); @@ -610,6 +838,7 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) } break; case UpdateSyncStatus syncStatus: + LastAppliedCheckpointRequestId = syncStatus.Status.LastAppliedCheckpointRequestId; UpdateSyncStatus(CoreInstructionHelpers.CoreStatusToSyncStatusOptions(syncStatus.Status)); break; case FetchCredentials fetchCredentials: @@ -655,7 +884,8 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) parameters = resolvedOptions.Params, active_streams = activeStreams, include_defaults = resolvedOptions.IncludeDefaultStreams, - app_metadata = resolvedOptions.AppMetadata + app_metadata = resolvedOptions.AppMetadata, + checkpoint_mode = resolvedOptions.CheckpointMode is CheckpointMode.Requests ? "requests" : "legacy", }; StreamingSyncRequest? establishRequest = null; @@ -673,18 +903,32 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) // "established" event is lost. commands = invocations.ListenAsync(nestedCts.Token); - // Wired up here rather than after this loop: a later instruction in this - // same batch (FetchCredentials) already needs to enqueue a command. - notifyCompletedUploads = () => + // Forwards upload completions for as long as this iteration lasts. One reported + // before the iteration started stays buffered in the channel, so it still + // reaches the core extension here. + _ = Task.Run(async () => { - if (!invocations.Closed) + try { - invocations.Emit(new EnqueuedCommand + while (!invocations.Closed) { - Command = PowerSyncControlCommand.NOTIFY_CRUD_UPLOAD_COMPLETED - }); + await crudUploadCompleted.Reader.ReadAsync(nestedCts.Token); + if (invocations.Closed) + { + return; + } + + invocations.Emit(new EnqueuedCommand + { + Command = PowerSyncControlCommand.NOTIFY_CRUD_UPLOAD_COMPLETED + }); + } } - }; + catch (OperationCanceledException) { /* Iteration ended. */ } + }); + + // Wired up here rather than after this loop: a later instruction in this + // same batch (FetchCredentials) already needs to enqueue a command. handleActiveStreamsChange = () => { if (!invocations.Closed) @@ -706,6 +950,28 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) }); } }; + + if (establish.CheckpointRequest is { } seedRequest) + { + // Run concurrently so that seeding checkpoint state doesn't block sync line processing. + seedingCheckpointState = Task.Run(async () => + { + try + { + await checkpointState.MarkCheckpointsReady( + () => SeedCheckpointRequestState(nestedCts.Token, seedRequest)); + } + catch (OperationCanceledException) { } + catch (Exception ex) + { + // Fail the download iteration if checkpoint requests are broken. + if (!invocations.Closed) + { + invocations.Emit(new EnqueuedCommand { Error = ex }); + } + } + }); + } } else if (startInstruction is CloseSyncStream) { @@ -729,6 +995,11 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) { await foreach (var command in commands!) { + if (command.Error != null) + { + System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(command.Error).Throw(); + } + var close = false; foreach (var instruction in await InvokePowerSyncControl(command.Command, command.Payload)) { @@ -739,7 +1010,7 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) if (instruction is CloseSyncStream closeSyncStream) { hideDisconnectOnRestart = closeSyncStream.HideDisconnect; - logger.LogWarning("Closing stream"); + logger.LogDebug("Closing stream"); close = true; break; } @@ -777,7 +1048,6 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) } finally { - notifyCompletedUploads = null; handleActiveStreamsChange = null; notifyTokenRefreshed = null; @@ -786,9 +1056,19 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) if (receivingLines != null) { - try { await receivingLines; } catch { /* surfaced via streamError */ } + try { await receivingLines; } catch { } + } + + // Let the seed settle before marking the iteration as ended, otherwise a seed completing + // during teardown could report readiness for an iteration that is already gone. + if (seedingCheckpointState != null) + { + try { await seedingCheckpointState; } catch { } } + // No checkpoint requests can be made until the next iteration seeds its state. + checkpointState.DownloadIterationEnded(); + await Stop(); } @@ -799,29 +1079,12 @@ public void Close() { crudUpdateCts?.Cancel(); crudUpdateCts = null; - try { crudUpdateTask?.Wait(2000); } catch (Exception) { } + try { crudUpdateTask?.Wait(2000); } catch { } Events.Close(); } - public record ResponseData( - [property: JsonProperty("write_checkpoint")] long WriteCheckpoint - ); - - public record ApiResponse( - [property: JsonProperty("data")] ResponseData Data - ); - public async Task GetWriteCheckpoint() + protected async Task InternalUploadAllCrud(CancellationToken signal, PowerSyncConnectionOptions options) { - var clientId = await Options.Adapter.GetClientId(); - var path = $"/write-checkpoint2.json?client_id={clientId}"; - var response = await Options.Remote.Get(path); - - return response.Data.WriteCheckpoint; - } - - protected async Task InternalUploadAllCrud() - { - await locks.ObtainLock(new LockOptions { Type = LockType.CRUD, @@ -829,16 +1092,16 @@ await locks.ObtainLock(new LockOptions { CrudEntry? checkedCrudItem = null; - while (true) + while (!signal.IsCancellationRequested) { - UpdateSyncStatus(new SyncStatusOptions { DataFlow = new SyncDataFlowStatus { Uploading = true } }); - try { // This is the first item in the FIFO CRUD queue. var nextCrudItem = await Options.Adapter.NextCrudItem(); if (nextCrudItem != null) { + UpdateSyncStatus(new SyncStatusOptions { DataFlow = new SyncDataFlowStatus { Uploading = true } }); + if (checkedCrudItem?.ClientId == nextCrudItem.ClientId) { logger.LogWarning( @@ -863,10 +1126,31 @@ await locks.ObtainLock(new LockOptions else { // Uploading is completed - await Options.Adapter.UpdateLocalTarget(GetWriteCheckpoint); + var neededUpdate = await Options.Adapter.UpdateLocalTarget(() => + options.CheckpointMode is CheckpointMode.Requests + ? RequestNextCheckpointFromService(signal) + : GetLegacyWriteCheckpoint(signal)); + if (!neededUpdate && checkedCrudItem != null) + { + // Only log this if there was something to upload + logger.LogDebug("Upload complete, no write checkpoint needed."); + } + UpdateSyncStatus(new SyncStatusOptions + { + DataFlow = new SyncDataFlowStatus + { + Uploading = false, + UploadError = null, + }, + }); break; } } + catch (OperationCanceledException) when (signal.IsCancellationRequested) + { + // Disconnecting. + break; + } catch (Exception ex) { checkedCrudItem = null; @@ -879,7 +1163,7 @@ await locks.ObtainLock(new LockOptions } }); - await DelayRetry(); + await DelayRetry(signal, options.RetryDelayMs ?? DEFAULT_RETRY_DELAY_MS); if (!IsConnected) { @@ -955,22 +1239,137 @@ protected void UpdateSyncStatus(SyncStatusOptions options, UpdateSyncStatusOptio } } - private async Task DelayRetry() + /// + /// Waits out a retry delay. Cancellation resolves the delay early instead of throwing an exception. + /// + /// + /// When set, the delay also ends as soon as a caller starts waiting to request a checkpoint. Such + /// a caller needs a seeded download iteration, so there is no point in making it wait out the + /// full delay. + /// + private async Task DelayRetry(CancellationToken signal, int delay, bool resumeOnCheckpointRequest = false) { - if (Options.RetryDelayMs.HasValue) + if (signal.IsCancellationRequested) { - await Task.Delay(Options.RetryDelayMs.Value); + return; + } + + using var nestedCts = CancellationTokenSource.CreateLinkedTokenSource(signal); + var timeout = Task.Delay(TimeSpan.FromMilliseconds(WithJitter(delay)), Options.TimeProvider, nestedCts.Token); + + if (resumeOnCheckpointRequest) + { + // WhenAny returns the winner without observing it, so neither branch throws here. + await Task.WhenAny(checkpointState.WaitForCheckpointWaiter(nestedCts.Token), timeout); + } + else + { + try + { + await timeout; + } + catch (OperationCanceledException) { /* Disconnecting. */ } } - } + // Ends whichever task is still pending to prevent abandoned tasks from consuming + // signals intended for future consumers of DelayRetry. + nestedCts.Cancel(); + } public void UpdateSubscriptions(SubscribedStream[] subscriptions) { activeStreams = subscriptions; handleActiveStreamsChange?.Invoke(); } + + /// + /// A conflating single-slot channel, equivalent to Channel.CONFLATED in Kotlin. + /// + private static Channel CreateNotifier() => Channel.CreateBounded(new BoundedChannelOptions(1) { FullMode = BoundedChannelFullMode.DropOldest }); + + private static readonly Random jitterRandom = new(); + + /// + /// Applies a random +/- 10% jitter to a delay, so that clients that dropped during the same + /// service outage don't all retry in lockstep. + /// + private static double WithJitter(int delayMs) + { + double factor; + lock (jitterRandom) + { + factor = 0.9 + (jitterRandom.NextDouble() * 0.2); + } + + return delayMs * factor; + } + + internal record LegacyWriteCheckpointResponseData( + [property: JsonProperty("write_checkpoint")] long WriteCheckpoint + ); + internal record LegacyWriteCheckpointApiResponse( + [property: JsonProperty("data")] LegacyWriteCheckpointResponseData Data + ); } +/// +/// The mechanism to request checkpoints from the PowerSync service. +/// +/// Checkpoint requests are used after a client uploads local mutations. The PowerSync service later references them in +/// downloaded data, allowing the SDK to assume that uploaded data has been synced down again. +/// +/// There are two ways to send checkpoint requests: A legacy (but default and stable) format supported by all PowerSync +/// service versions, and a newer (`requests`) method which is only available from PowerSync service version 1.24.0 or +/// later. +/// +/// Note that the requests checkpoint mode is an alpha API. +/// +public record CheckpointMode +{ + private CheckpointMode() { } + + /// + /// Uses a legacy but stable endpoint to request checkpoints. + /// + public static readonly CheckpointMode Legacy = new(); + + /// + /// Adopts a new and more efficient checkpoint protocol with better support for switching users + /// on devices. + /// + public sealed record Requests : CheckpointMode + { + const int MINIMUM_RETRY_DELAY = 10_000; + const int DEFAULT_RETRY_DELAY = MINIMUM_RETRY_DELAY; + + /// + /// The periodic interval before re-posting the latest checkpoint request to the service if + /// it has not been applied in time. + /// + public int RetryDelayMs { get; } + + /// + /// Use checkpoint requests with the default retry delay. + /// + public Requests() + { + RetryDelayMs = DEFAULT_RETRY_DELAY; + } + + /// + /// Use checkpoint requests with a custom retry delay. + /// + /// Thrown when retry delay is less than + public Requests(int retryDelayMs) + { + if (retryDelayMs < MINIMUM_RETRY_DELAY) + { + throw new ArgumentException($"Retry delay must be at least {MINIMUM_RETRY_DELAY}ms."); + } + RetryDelayMs = retryDelayMs; + } + } +} enum LockType { diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs index 3f1d6ec2..71694c77 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs @@ -1,14 +1,14 @@ -namespace PowerSync.Common.Client.Sync.Stream; +using PowerSync.Common.Client.Sync.Bucket; +using PowerSync.Common.DB.Crud; using Newtonsoft.Json; -using PowerSync.Common.Client.Sync.Bucket; -using PowerSync.Common.DB.Crud; +namespace PowerSync.Common.Client.Sync.Stream; public class ContinueCheckpointRequest { [JsonProperty("buckets")] - public List Buckets { get; set; } = new(); + public List Buckets { get; set; } = []; [JsonProperty("checkpoint_token")] public string CheckpointToken { get; set; } = ""; @@ -95,7 +95,7 @@ public class RequestStreamSubscription public string Stream { get; set; } = ""; [JsonProperty("parameters")] - public Dictionary Parameters { get; set; } = new(); + public Dictionary Parameters { get; set; } = []; [JsonProperty("override_priority")] public int? OverridePriority { get; set; } @@ -128,16 +128,16 @@ public class StreamingSyncCheckpointDiff : StreamingSyncLine public class CheckpointDiff { [JsonProperty("last_op_id")] - public string LastOpId { get; set; } = ""; + public long LastOpId { get; set; } [JsonProperty("updated_buckets")] - public List UpdatedBuckets { get; set; } = new(); + public List UpdatedBuckets { get; set; } = []; [JsonProperty("removed_buckets")] - public List RemovedBuckets { get; set; } = new(); + public List RemovedBuckets { get; set; } = []; [JsonProperty("write_checkpoint")] - public string WriteCheckpoint { get; set; } = ""; + public long WriteCheckpoint { get; set; } } public class StreamingSyncDataJSON : StreamingSyncLine @@ -182,7 +182,7 @@ public class StreamingSyncKeepalive : StreamingSyncLine public class CrudRequest { [JsonProperty("data")] - public List Data { get; set; } = new(); + public List Data { get; set; } = []; } public class CrudResponse @@ -190,3 +190,24 @@ public class CrudResponse [JsonProperty("checkpoint")] public string? Checkpoint { get; set; } } + +public class CheckpointRequestPayload +{ + [JsonProperty("client_id")] + public string ClientId { get; set; } = ""; + + [JsonProperty("checkpoint_request_id")] + public long CheckpointRequestId { get; set; } +} + +public class CheckpointRequestResponse +{ + [JsonProperty("data")] + public CheckpointRequestResponseData Data { get; set; } = new(); +} + +public class CheckpointRequestResponseData +{ + [JsonProperty("checkpoint_request_id")] + public long CheckpointRequestId { get; set; } +} diff --git a/PowerSync/PowerSync.Common/DB/Crud/SyncProgress.cs b/PowerSync/PowerSync.Common/DB/Crud/SyncProgress.cs index 7bab6911..8aefc5ff 100644 --- a/PowerSync/PowerSync.Common/DB/Crud/SyncProgress.cs +++ b/PowerSync/PowerSync.Common/DB/Crud/SyncProgress.cs @@ -20,11 +20,11 @@ namespace PowerSync.Common.DB.Crud; public class SyncProgress : ProgressWithOperations { public static readonly int FULL_SYNC_PRIORITY = 2147483647; - protected Dictionary InternalProgress { get; } + private Dictionary InternalProgress { get; } - public SyncProgress(Dictionary progress) + internal SyncProgress(Dictionary progress) { - this.InternalProgress = progress; + InternalProgress = progress; var untilCompletion = UntilPriority(FULL_SYNC_PRIORITY); TotalOperations = untilCompletion.TotalOperations; diff --git a/PowerSync/PowerSync.Common/DB/Crud/SyncStatus.cs b/PowerSync/PowerSync.Common/DB/Crud/SyncStatus.cs index 2bd94213..78b6bcbe 100644 --- a/PowerSync/PowerSync.Common/DB/Crud/SyncStatus.cs +++ b/PowerSync/PowerSync.Common/DB/Crud/SyncStatus.cs @@ -1,7 +1,5 @@ namespace PowerSync.Common.DB.Crud; -using Microsoft.Extensions.Options; - using Newtonsoft.Json; using PowerSync.Common.Client.Sync.Stream; @@ -47,9 +45,7 @@ public class SyncPriorityStatus public class SyncStatusOptions { - public SyncStatusOptions() - { - } + public SyncStatusOptions() { } public SyncStatusOptions(SyncStatusOptions options) { @@ -221,7 +217,6 @@ public SyncStreamStatusView[]? SyncStreams return raw != null ? new SyncStreamStatusView(this, raw) : null; } - private string SerializeObject() { return JsonConvert.SerializeObject(new { Options, UploadErrorMessage = Options.DataFlow?.UploadError?.Message, DownloadErrorMessage = DataFlowStatus.DownloadError?.Message }); diff --git a/PowerSync/PowerSync.Common/PowerSync.Common.csproj b/PowerSync/PowerSync.Common/PowerSync.Common.csproj index 6e45bb7d..01794909 100644 --- a/PowerSync/PowerSync.Common/PowerSync.Common.csproj +++ b/PowerSync/PowerSync.Common/PowerSync.Common.csproj @@ -1,7 +1,9 @@  - netstandard2.0;net6.0;net8.0;net9.0;net8.0-ios;net8.0-android;net8.0-maccatalyst;net9.0-ios;net9.0-android;net9.0-maccatalyst + net8.0;net9.0;net10.0;net9.0-android;net10.0-android + $(TargetFrameworks);net9.0-ios;net10.0-ios;net9.0-maccatalyst;net10.0-maccatalyst + 12 enable enable @@ -17,8 +19,10 @@ https://github.com/powersync-ja/powersync-dotnet/PowerSync/PowerSync.Common/CHANGELOG.md powersync local-first local-storage state-management offline sql db persistence sqlite sync icon.png - NU5100 + NU5100;1591 + true README.md + true $(DefaultItemExcludes);runtimes/**/*.*; @@ -27,21 +31,19 @@ - - - - - - + + + + + - - + diff --git a/PowerSync/PowerSync.Common/Utils/BroadcastChannel.cs b/PowerSync/PowerSync.Common/Utils/BroadcastChannel.cs new file mode 100644 index 00000000..92a4bc97 --- /dev/null +++ b/PowerSync/PowerSync.Common/Utils/BroadcastChannel.cs @@ -0,0 +1,47 @@ +using System.Threading.Channels; +using System.Collections.Concurrent; + +namespace PowerSync.Common.Utils; + +/// +/// -like object that allows multiple listeners at once and +/// broadcasts messages to all subscribers instead of sending any given message to +/// exactly one consumer. +/// +internal class BroadcastChannel +{ + private readonly ConcurrentDictionary> _subscribers = new(); + + public ChannelReader Subscribe(out Guid subscriberId) + { + subscriberId = Guid.NewGuid(); + var ch = Channel.CreateUnbounded(); + _subscribers.TryAdd(subscriberId, ch.Writer); + return ch.Reader; + } + + public void Unsubscribe(Guid id) + { + if (_subscribers.TryRemove(id, out var writer)) + { + writer.Complete(); + } + } + + public void Broadcast(T message) + { + foreach (ChannelWriter writer in _subscribers.Values) + { + writer.TryWrite(message); + } + } + + public async Task BroadcastAsync(T message) + { + foreach (ChannelWriter writer in _subscribers.Values) + { + await writer.WriteAsync(message); + } + } +} + diff --git a/PowerSync/PowerSync.Common/Utils/Converters/StringLongConverter.cs b/PowerSync/PowerSync.Common/Utils/Converters/StringLongConverter.cs new file mode 100644 index 00000000..61c7cbdf --- /dev/null +++ b/PowerSync/PowerSync.Common/Utils/Converters/StringLongConverter.cs @@ -0,0 +1,47 @@ +using Newtonsoft.Json; + +namespace PowerSync.Common.Utils.Converters; + +/// +/// Converts a long to and from a string when converting JSON values. Used +/// for converting checkpoint request IDs from a long to a string before being +/// passed to the core extension. +/// +/// TODO: This is not currently in use because checkpoint request IDs are +/// currently represented as strings, however this is going to change +/// in the 1.0 release. +/// +internal class StringLongConverter : JsonConverter +{ + public override bool CanConvert(Type objectType) + { + return objectType == typeof(long) || objectType == typeof(long?); + } + + public override void WriteJson(JsonWriter writer, object? value, JsonSerializer serializer) + { + if (value == null) + { + writer.WriteNull(); + } + else + { + writer.WriteValue(value.ToString()); + } + } + + public override object ReadJson(JsonReader reader, Type objectType, object? existingValue, JsonSerializer serializer) + { + if (reader.TokenType == JsonToken.Null) + return null!; + + var val = reader.Value?.ToString(); + + if (long.TryParse(val, out long result)) + { + return result; + } + + throw new JsonSerializationException($"Cannot convert value {val} to long."); + } +} diff --git a/PowerSync/PowerSync.Maui/CHANGELOG.md b/PowerSync/PowerSync.Maui/CHANGELOG.md index 09ce912c..72f1d515 100644 --- a/PowerSync/PowerSync.Maui/CHANGELOG.md +++ b/PowerSync/PowerSync.Maui/CHANGELOG.md @@ -9,6 +9,10 @@ - Accidental release (mirror of 0.1.2). Use 1.0.1 instead. +## 0.1.5 + +- Upstream PowerSync.Common version bump (See Powersync.Common changelog 0.1.5 for more information) + ## 0.1.4 - Upstream PowerSync.Common version bump (See Powersync.Common changelog 0.1.4 for more information) diff --git a/PowerSync/PowerSync.Maui/PowerSync.Maui.csproj b/PowerSync/PowerSync.Maui/PowerSync.Maui.csproj index 74e9a948..f8864efb 100644 --- a/PowerSync/PowerSync.Maui/PowerSync.Maui.csproj +++ b/PowerSync/PowerSync.Maui/PowerSync.Maui.csproj @@ -1,7 +1,9 @@  - netstandard2.0;net6.0;net8.0;net9.0;net8.0-ios;net8.0-android;net8.0-maccatalyst;net9.0-ios;net9.0-android;net9.0-maccatalyst + net9.0;net10.0;net9.0-android;net10.0-android + $(TargetFrameworks);net9.0-ios;net10.0-ios;net9.0-maccatalyst;net10.0-maccatalyst + 12 enable enable @@ -17,8 +19,9 @@ https://github.com/powersync-ja/powersync-dotnet/PowerSync/PowerSync.Maui/CHANGELOG.md powersync local-first local-storage state-management offline sql db persistence sqlite sync icon.png - NU5100 + NU5100;1591 README.md + true true true diff --git a/README.md b/README.md index b22b0a27..a48185b4 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@

-_[PowerSync](https://www.powersync.com) is a sync engine for building local-first apps with instantly-responsive UI/UX and simplified state transfer. Syncs between SQLite on the client-side and Postgres, MongoDB, MySQL or SQL Server on the server-side._ +_[PowerSync](https://www.powersync.com) keeps a client-side SQLite database in sync with your backend database. Changes appear across users and devices in real-time, user interactions feel instant and your app continues to work even when offline. Supports Postgres, MongoDB, MySQL, and SQL Server. Client SDKs are available for a wide range of environments including web, mobile, desktop, headless and embedded._ # PowerSync .NET SDKs @@ -110,13 +110,13 @@ dotnet restore Run all tests ```bash -dotnet test -v n --framework net8.0 +dotnet test -v n --framework net10.0 ``` Run a specific test ```bash -dotnet test -v n --framework net8.0 --filter "test-file-pattern" +dotnet test -v n --framework net10.0 --filter "test-file-pattern" ``` ### Integration Tests @@ -125,13 +125,13 @@ Integration tests in `PowerSync.Common.IntegrationTests` are intended to run aga The integration tests are disabled by default, define the following environment variable when running the tests. ```bash -RUN_INTEGRATION_TESTS=true dotnet test -v n --framework net8.0 +RUN_INTEGRATION_TESTS=true dotnet test -v n --framework net10.0 ``` Only run integration tests, without any unit tests. ```bash -RUN_INTEGRATION_TESTS=true dotnet test -v n --framework net8.0 --filter "Category=Integration" +RUN_INTEGRATION_TESTS=true dotnet test -v n --framework net10.0 --filter "Category=Integration" ``` ### Performance Tests @@ -139,13 +139,13 @@ RUN_INTEGRATION_TESTS=true dotnet test -v n --framework net8.0 --filter "Categor Performance tests in `PowerSync.Common.PerformanceTests` are disabled by default, define the following environment variable when running the tests. ```bash -RUN_PERFORMANCE_TESTS=true dotnet test -v n --framework net8.0 +RUN_PERFORMANCE_TESTS=true dotnet test -v n --framework net10.0 ``` Only run performance tests, without any unit tests. ```bash -RUN_PERFORMANCE_TESTS=true dotnet test -v n --framework net8.0 --filter "Category=Performance" +RUN_PERFORMANCE_TESTS=true dotnet test -v n --framework net10.0 --filter "Category=Performance" ``` ## Using the PowerSync.Common package in your project diff --git a/Tests/PowerSync/PowerSync.Common.IntegrationTests/NodeClient.cs b/Tests/PowerSync/PowerSync.Common.IntegrationTests/NodeClient.cs index dbd5073e..5ce805b7 100644 --- a/Tests/PowerSync/PowerSync.Common.IntegrationTests/NodeClient.cs +++ b/Tests/PowerSync/PowerSync.Common.IntegrationTests/NodeClient.cs @@ -1,161 +1,161 @@ -namespace PowerSync.Common.IntegrationTests; - -using System; -using System.Collections.Generic; -using System.Net.Http; -using System.Text; -using System.Text.Json; -using System.Threading.Tasks; - -using PowerSync.Common.DB.Crud; - -public class NodeClient -{ - private readonly HttpClient _httpClient; - private readonly string _backendUrl; - private readonly string _userId; - - public NodeClient(string userId) - { - _httpClient = new HttpClient(); - _backendUrl = "http://localhost:6060"; - _userId = userId; - } - - public NodeClient(string backendUrl, string userId) - { - _httpClient = new HttpClient(); - _backendUrl = backendUrl; - _userId = userId; - } - - public Task CreateList(string id, string name) - { - return CreateItem("lists", id, name); - } - - async Task CreateItem(string table, string id, string name) - { - var data = new Dictionary - { - { "created_at", DateTime.UtcNow.ToString("yyyy-MM-dd HH:mm:ss") }, - { "name", name }, - { "owner_id", _userId } - }; - - var batch = new[] - { - new - { - op = UpdateType.PUT.ToString(), - table = table, - id = id, - data = data - } - }; - - var payload = JsonSerializer.Serialize(new { batch }); - var content = new StringContent(payload, Encoding.UTF8, "application/json"); - - HttpResponseMessage response = await _httpClient.PostAsync($"{_backendUrl}/api/data", content); - - if (!response.IsSuccessStatusCode) - { - Console.WriteLine(await response.Content.ReadAsStringAsync()); - throw new Exception( - $"Failed to create item. Status: {response.StatusCode}, " + - $"Response: {await response.Content.ReadAsStringAsync()}" - ); - } - - return await response.Content.ReadAsStringAsync(); - } - - public Task DeleteList(string id) - { - return DeleteItem("lists", id); - } - - public Task CreateTodo(string id, string listId, string description) - { - return CreateTodoItem("todos", id, listId, description); - } - - async Task CreateTodoItem(string table, string id, string listId, string description) - { - var data = new Dictionary - { - { "created_at", DateTime.UtcNow.ToString("yyyy-MM-dd HH:mm:ss") }, - { "description", description }, - { "list_id", listId }, - { "created_by", _userId }, - { "completed", 0 }, - }; - - var batch = new[] - { - new - { - op = UpdateType.PUT.ToString(), - table = table, - id = id, - data = data - } - }; - - var payload = JsonSerializer.Serialize(new { batch }); - var content = new StringContent(payload, Encoding.UTF8, "application/json"); - - HttpResponseMessage response = await _httpClient.PostAsync($"{_backendUrl}/api/data", content); - - if (!response.IsSuccessStatusCode) - { - Console.WriteLine(await response.Content.ReadAsStringAsync()); - throw new Exception( - $"Failed to create todo. Status: {response.StatusCode}, " + - $"Response: {await response.Content.ReadAsStringAsync()}" - ); - } - - return await response.Content.ReadAsStringAsync(); - } - - public Task DeleteTodo(string id) - { - return DeleteItem("todos", id); - } - - async Task DeleteItem(string table, string id) - { - var batch = new[] - { - new - { - op = UpdateType.DELETE.ToString(), - table = table, - id = id - } - }; - - var payload = JsonSerializer.Serialize(new { batch }); - var content = new StringContent(payload, Encoding.UTF8, "application/json"); - - HttpResponseMessage response = await _httpClient.PostAsync($"{_backendUrl}/api/data", content); - - if (!response.IsSuccessStatusCode) - { - Console.WriteLine(await response.Content.ReadAsStringAsync()); - throw new Exception( - $"Failed to delete item. Status: {response.StatusCode}, " + - $"Response: {await response.Content.ReadAsStringAsync()}" - ); - } - - return await response.Content.ReadAsStringAsync(); - } - - public void Dispose() - { - _httpClient?.Dispose(); - } +namespace PowerSync.Common.IntegrationTests; + +using System; +using System.Collections.Generic; +using System.Net.Http; +using System.Text; +using System.Text.Json; +using System.Threading.Tasks; + +using PowerSync.Common.DB.Crud; + +public class NodeClient +{ + private readonly HttpClient _httpClient; + private readonly string _backendUrl; + private readonly string _userId; + + public NodeClient(string userId) + { + _httpClient = new HttpClient(); + _backendUrl = "http://localhost:6060"; + _userId = userId; + } + + public NodeClient(string backendUrl, string userId) + { + _httpClient = new HttpClient(); + _backendUrl = backendUrl; + _userId = userId; + } + + public Task CreateList(string id, string name) + { + return CreateItem("lists", id, name); + } + + async Task CreateItem(string table, string id, string name) + { + var data = new Dictionary + { + { "created_at", DateTime.UtcNow.ToString("yyyy-MM-dd HH:mm:ss") }, + { "name", name }, + { "owner_id", _userId } + }; + + var batch = new[] + { + new + { + op = UpdateType.PUT.ToString(), + table = table, + id = id, + data = data + } + }; + + var payload = JsonSerializer.Serialize(new { batch }); + var content = new StringContent(payload, Encoding.UTF8, "application/json"); + + HttpResponseMessage response = await _httpClient.PostAsync($"{_backendUrl}/api/data", content); + + if (!response.IsSuccessStatusCode) + { + Console.WriteLine(await response.Content.ReadAsStringAsync()); + throw new Exception( + $"Failed to create item. Status: {response.StatusCode}, " + + $"Response: {await response.Content.ReadAsStringAsync()}" + ); + } + + return await response.Content.ReadAsStringAsync(); + } + + public Task DeleteList(string id) + { + return DeleteItem("lists", id); + } + + public Task CreateTodo(string id, string listId, string description) + { + return CreateTodoItem("todos", id, listId, description); + } + + async Task CreateTodoItem(string table, string id, string listId, string description) + { + var data = new Dictionary + { + { "created_at", DateTime.UtcNow.ToString("yyyy-MM-dd HH:mm:ss") }, + { "description", description }, + { "list_id", listId }, + { "created_by", _userId }, + { "completed", 0 }, + }; + + var batch = new[] + { + new + { + op = UpdateType.PUT.ToString(), + table = table, + id = id, + data = data + } + }; + + var payload = JsonSerializer.Serialize(new { batch }); + var content = new StringContent(payload, Encoding.UTF8, "application/json"); + + HttpResponseMessage response = await _httpClient.PostAsync($"{_backendUrl}/api/data", content); + + if (!response.IsSuccessStatusCode) + { + Console.WriteLine(await response.Content.ReadAsStringAsync()); + throw new Exception( + $"Failed to create todo. Status: {response.StatusCode}, " + + $"Response: {await response.Content.ReadAsStringAsync()}" + ); + } + + return await response.Content.ReadAsStringAsync(); + } + + public Task DeleteTodo(string id) + { + return DeleteItem("todos", id); + } + + async Task DeleteItem(string table, string id) + { + var batch = new[] + { + new + { + op = UpdateType.DELETE.ToString(), + table = table, + id = id + } + }; + + var payload = JsonSerializer.Serialize(new { batch }); + var content = new StringContent(payload, Encoding.UTF8, "application/json"); + + HttpResponseMessage response = await _httpClient.PostAsync($"{_backendUrl}/api/data", content); + + if (!response.IsSuccessStatusCode) + { + Console.WriteLine(await response.Content.ReadAsStringAsync()); + throw new Exception( + $"Failed to delete item. Status: {response.StatusCode}, " + + $"Response: {await response.Content.ReadAsStringAsync()}" + ); + } + + return await response.Content.ReadAsStringAsync(); + } + + public void Dispose() + { + _httpClient?.Dispose(); + } } diff --git a/Tests/PowerSync/PowerSync.Common.IntegrationTests/NodeConnector.cs b/Tests/PowerSync/PowerSync.Common.IntegrationTests/NodeConnector.cs index ae2a1b2b..6585a42b 100644 --- a/Tests/PowerSync/PowerSync.Common.IntegrationTests/NodeConnector.cs +++ b/Tests/PowerSync/PowerSync.Common.IntegrationTests/NodeConnector.cs @@ -1,114 +1,114 @@ -namespace PowerSync.Common.IntegrationTests; - - -using System; -using System.Collections.Generic; -using System.IO; -using System.Net.Http; -using System.Text; -using System.Text.Json; -using System.Threading.Tasks; - -using PowerSync.Common.Client; -using PowerSync.Common.Client.Connection; -using PowerSync.Common.DB.Crud; - - -public class NodeConnector : IPowerSyncBackendConnector -{ - private readonly HttpClient _httpClient; - - public string BackendUrl { get; } - public string PowerSyncUrl { get; } - public string UserId { get; private set; } - private string? clientId; - - public NodeConnector(string userId) - { - _httpClient = new HttpClient(); - - // Load or generate User ID - UserId = userId; - - BackendUrl = "http://localhost:6060"; - PowerSyncUrl = "http://localhost:8080"; - - clientId = null; - } - - public async Task FetchCredentials() - { - string tokenEndpoint = "api/auth/token"; - string url = $"{BackendUrl}/{tokenEndpoint}?user_id={UserId}"; - - HttpResponseMessage response = await _httpClient.GetAsync(url); - if (!response.IsSuccessStatusCode) - { - throw new Exception($"Received {response.StatusCode} from {tokenEndpoint}: {await response.Content.ReadAsStringAsync()}"); - } - - string responseBody = await response.Content.ReadAsStringAsync(); - var jsonResponse = JsonSerializer.Deserialize>(responseBody); - - if (jsonResponse == null || !jsonResponse.ContainsKey("token")) - { - throw new Exception("Invalid response received from authentication endpoint."); - } - - return new PowerSyncCredentials(PowerSyncUrl, jsonResponse["token"]); - } - - public async Task UploadData(IPowerSyncDatabase database) - { - CrudTransaction? transaction; - try - { - transaction = await database.GetNextCrudTransaction(); - } - catch (Exception ex) - { - Console.WriteLine($"UploadData Error: {ex.Message}"); - return; - } - - if (transaction == null) - { - return; - } - - clientId ??= await database.GetClientId(); - - try - { - var batch = new List(); - - foreach (var operation in transaction.Crud) - { - batch.Add(new - { - op = operation.Op.ToString(), - table = operation.Table, - id = operation.Id, - data = operation.OpData - }); - } - - var payload = JsonSerializer.Serialize(new { batch }); - var content = new StringContent(payload, Encoding.UTF8, "application/json"); - - HttpResponseMessage response = await _httpClient.PostAsync($"{BackendUrl}/api/data", content); - - if (!response.IsSuccessStatusCode) - { - throw new Exception($"Received {response.StatusCode} from /api/data: {await response.Content.ReadAsStringAsync()}"); - } - - await transaction.Complete(); - } - catch (Exception ex) - { - Console.WriteLine($"UploadData Error: {ex.Message}"); - throw; - } - } +namespace PowerSync.Common.IntegrationTests; + + +using System; +using System.Collections.Generic; +using System.IO; +using System.Net.Http; +using System.Text; +using System.Text.Json; +using System.Threading.Tasks; + +using PowerSync.Common.Client; +using PowerSync.Common.Client.Connection; +using PowerSync.Common.DB.Crud; + + +public class NodeConnector : IPowerSyncBackendConnector +{ + private readonly HttpClient _httpClient; + + public string BackendUrl { get; } + public string PowerSyncUrl { get; } + public string UserId { get; private set; } + private string? clientId; + + public NodeConnector(string userId) + { + _httpClient = new HttpClient(); + + // Load or generate User ID + UserId = userId; + + BackendUrl = "http://localhost:6060"; + PowerSyncUrl = "http://localhost:8080"; + + clientId = null; + } + + public async Task FetchCredentials() + { + string tokenEndpoint = "api/auth/token"; + string url = $"{BackendUrl}/{tokenEndpoint}?user_id={UserId}"; + + HttpResponseMessage response = await _httpClient.GetAsync(url); + if (!response.IsSuccessStatusCode) + { + throw new Exception($"Received {response.StatusCode} from {tokenEndpoint}: {await response.Content.ReadAsStringAsync()}"); + } + + string responseBody = await response.Content.ReadAsStringAsync(); + var jsonResponse = JsonSerializer.Deserialize>(responseBody); + + if (jsonResponse == null || !jsonResponse.ContainsKey("token")) + { + throw new Exception("Invalid response received from authentication endpoint."); + } + + return new PowerSyncCredentials(PowerSyncUrl, jsonResponse["token"]); + } + + public async Task UploadData(IPowerSyncDatabase database) + { + CrudTransaction? transaction; + try + { + transaction = await database.GetNextCrudTransaction(); + } + catch (Exception ex) + { + Console.WriteLine($"UploadData Error: {ex.Message}"); + return; + } + + if (transaction == null) + { + return; + } + + clientId ??= await database.GetClientId(); + + try + { + var batch = new List(); + + foreach (var operation in transaction.Crud) + { + batch.Add(new + { + op = operation.Op.ToString(), + table = operation.Table, + id = operation.Id, + data = operation.OpData + }); + } + + var payload = JsonSerializer.Serialize(new { batch }); + var content = new StringContent(payload, Encoding.UTF8, "application/json"); + + HttpResponseMessage response = await _httpClient.PostAsync($"{BackendUrl}/api/data", content); + + if (!response.IsSuccessStatusCode) + { + throw new Exception($"Received {response.StatusCode} from /api/data: {await response.Content.ReadAsStringAsync()}"); + } + + await transaction.Complete(); + } + catch (Exception ex) + { + Console.WriteLine($"UploadData Error: {ex.Message}"); + throw; + } + } } diff --git a/Tests/PowerSync/PowerSync.Common.IntegrationTests/PowerSync.Common.IntegrationTests.csproj b/Tests/PowerSync/PowerSync.Common.IntegrationTests/PowerSync.Common.IntegrationTests.csproj index 3d029a9c..eefd14e9 100644 --- a/Tests/PowerSync/PowerSync.Common.IntegrationTests/PowerSync.Common.IntegrationTests.csproj +++ b/Tests/PowerSync/PowerSync.Common.IntegrationTests/PowerSync.Common.IntegrationTests.csproj @@ -1,7 +1,7 @@ - + - net8.0 + net8.0;net9.0;net10.0 12 enable enable diff --git a/Tests/PowerSync/PowerSync.Common.IntegrationTests/SyncIntegrationTests.cs b/Tests/PowerSync/PowerSync.Common.IntegrationTests/SyncIntegrationTests.cs index 699d9e41..4ee7c4d8 100644 --- a/Tests/PowerSync/PowerSync.Common.IntegrationTests/SyncIntegrationTests.cs +++ b/Tests/PowerSync/PowerSync.Common.IntegrationTests/SyncIntegrationTests.cs @@ -9,9 +9,9 @@ namespace PowerSync.Common.IntegrationTests; [Trait("Category", "Integration")] public class SyncIntegrationTests : IAsyncLifetime { - private record ListResult(string id, string name, string owner_id, string created_at); + private record ListResult(string id, string created_at, string name, string owner_id); - private record TodoResult(string id, string list_id, string content, string owner_id, string created_at); + private record TodoResult(string id, string list_id, string created_at, string completed_at, string description, string created_by, string completed_by, int completed); private readonly string userId = Uuid(); @@ -33,12 +33,13 @@ public async Task InitializeAsync() nodeClient = new NodeClient(userId); db = new PowerSyncDatabase(new PowerSyncDatabaseOptions { - Database = new SQLOpenOptions { DbFilename = "powersync-sync-tests.db" }, + Database = new SQLOpenOptions { DbFilename = $"powersync-sync-tests-{userId}.db" }, Schema = TestSchema.PowerSyncSchema, Logger = logger }); await db.Init(); + await db.DisconnectAndClear(); var connector = new NodeConnector(userId); Console.WriteLine($"Using User ID: {userId}"); @@ -52,7 +53,6 @@ public async Task InitializeAsync() { "environment", "integration-tests" } } }); - await db.Connect(connector); await db.WaitForFirstSync(); } catch (Exception ex) @@ -68,6 +68,7 @@ public async Task DisposeAsync() await Task.Delay(2000); await db.DisconnectAndClear(); await db.Close(); + try { File.Delete($"powersync-sync-tests-{userId}.db"); } catch { } } [IntegrationFact(Timeout = 3000)] @@ -93,48 +94,39 @@ public async Task SyncDownCreateOperationTest() await watched.Task; } - [IntegrationFact(Timeout = 3000)] + [IntegrationFact(Timeout = 6000)] public async Task SyncDownDeleteOperationTest() { - var watched = new TaskCompletionSource(); + var created = new TaskCompletionSource(); + var deleted = new TaskCompletionSource(); var cts = new CancellationTokenSource(); var id = Uuid(); - await nodeClient.CreateList(id, name: "Test List to delete"); - + // Use a single Watch for the full create+delete lifecycle so the listener + // channel is registered before any events can be missed. _ = Task.Run(async () => { + bool sawCreate = false; await foreach (var x in db.Watch("select * from lists where id = ?", [id], new() { Signal = cts.Token })) { - // Verify that the item was added locally - if (x.Length == 1) + if (!sawCreate && x.Length == 1) { - watched.SetResult(true); - cts.Cancel(); + sawCreate = true; + created.SetResult(true); } - } - }); - - await watched.Task; - await nodeClient.DeleteList(id); - - watched = new TaskCompletionSource(); - cts = new CancellationTokenSource(); - - _ = Task.Run(async () => - { - await foreach (var x in db.Watch("select * from lists where id = ?", [id], new() { Signal = cts.Token })) - { - // Verify that the item was deleted locally - if (x.Length == 0) + else if (sawCreate && x.Length == 0) { - watched.SetResult(true); + deleted.SetResult(true); cts.Cancel(); } } }); - await watched.Task; + await nodeClient.CreateList(id, name: "Test List to delete"); + await created.Task; + + await nodeClient.DeleteList(id); + await deleted.Task; } [IntegrationFact(Timeout = 5000)] @@ -147,7 +139,7 @@ public async Task SyncDownLargeCreateOperationTest() _ = Task.Run(async () => { - await foreach (var x in db.Watch("select * from lists where id = ?", [id], new() { Signal = cts.Token })) + await foreach (var x in db.Watch("select * from lists where name = ?", [listName], new() { Signal = cts.Token })) { // Verify that the item was added locally if (x.Length == 100) @@ -176,7 +168,7 @@ public async Task SyncDownCreateOperationAfterLargeUploadTest() _ = Task.Run(async () => { - await foreach (var x in db.Watch("select * from lists where id = ?", [id], new() { Signal = cts.Token })) + await foreach (var x in db.Watch("select * from lists where name = ?", [listName], new() { Signal = cts.Token })) { // Verify that the items were added locally if (x.Length == 100) @@ -199,8 +191,11 @@ await db.Execute("insert into lists (id, name, owner_id, created_at) values (uui } await localInsertWatch.Task; - // let the crud upload finish - await Task.Delay(2000); + // Wait for CRUD upload to complete before creating backend item + while (db.CurrentStatus.DataFlowStatus.Uploading) + { + await Task.Delay(50); + } await nodeClient.CreateList(Uuid(), listName); await backendInsertWatch.Task; diff --git a/Tests/PowerSync/PowerSync.Common.IntegrationTests/xunit.runner.json b/Tests/PowerSync/PowerSync.Common.IntegrationTests/xunit.runner.json index fcdb064a..27dcc91d 100644 --- a/Tests/PowerSync/PowerSync.Common.IntegrationTests/xunit.runner.json +++ b/Tests/PowerSync/PowerSync.Common.IntegrationTests/xunit.runner.json @@ -1,5 +1,7 @@ { "methodDisplay": "method", "diagnosticMessages": true, - "longRunningTestSeconds": 10 -} \ No newline at end of file + "longRunningTestSeconds": 10, + "parallelizeTestCollections": false, + "maxParallelThreads": 1 +} diff --git a/Tests/PowerSync/PowerSync.Common.PerformanceTests/PowerSync.Common.PerformanceTests.csproj b/Tests/PowerSync/PowerSync.Common.PerformanceTests/PowerSync.Common.PerformanceTests.csproj index 25d9697b..7773caa6 100644 --- a/Tests/PowerSync/PowerSync.Common.PerformanceTests/PowerSync.Common.PerformanceTests.csproj +++ b/Tests/PowerSync/PowerSync.Common.PerformanceTests/PowerSync.Common.PerformanceTests.csproj @@ -1,7 +1,7 @@  - net6.0;net8.0;net9.0 + net8.0;net9.0;net10.0 12 enable enable @@ -11,10 +11,16 @@ - - - - + + runtime; build; native; contentfiles; analyzers; buildtransitive + all + + + + + runtime; build; native; contentfiles; analyzers; buildtransitive + all + diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Attachments/AttachmentTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Attachments/AttachmentTests.cs index af883348..9dfc49d3 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Attachments/AttachmentTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Attachments/AttachmentTests.cs @@ -8,7 +8,7 @@ namespace PowerSync.Common.Tests.Attachments; using PowerSync.Common.Tests.Utils; /// -/// dotnet test -v n --framework net8.0 --filter "AttachmentTests" +/// dotnet test -v n --framework net10.0 --filter "AttachmentTests" /// [Collection("AttachmentTests")] public class AttachmentTests : IAsyncLifetime diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/PowerSyncDatabaseTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/PowerSyncDatabaseTests.cs index 19e40f03..56cebe4c 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/PowerSyncDatabaseTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/PowerSyncDatabaseTests.cs @@ -11,7 +11,7 @@ namespace PowerSync.Common.Tests.Client; using PowerSync.Common.Tests.Utils; /// -/// dotnet test -v n --framework net8.0 --filter "PowerSyncDatabaseTests" +/// dotnet test -v n --framework net10.0 --filter "PowerSyncDatabaseTests" /// [Collection("PowerSyncDatabaseTests")] public class PowerSyncDatabaseTests : IAsyncLifetime @@ -1096,7 +1096,7 @@ await db.UpdateSchema(new Schema( Assert.True(await sem.WaitAsync(500)); Assert.Single(events); Assert.True(events.TryDequeue(out var change)); - Assert.Equal(["assets"], change.ChangedTables); + Assert.Equal(["ps_data__assets"], change.ChangedTables); } [Fact] diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CRUDTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CRUDTests.cs index 58525fb5..094a0921 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CRUDTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CRUDTests.cs @@ -11,7 +11,7 @@ namespace PowerSync.Common.Tests.Client.Sync; /// -/// dotnet test -v n --framework net8.0 --filter "CRUDTests" +/// dotnet test -v n --framework net10.0 --filter "CRUDTests" /// public class CRUDTests : IAsyncLifetime { diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs new file mode 100644 index 00000000..01227bfb --- /dev/null +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs @@ -0,0 +1,685 @@ +using Microsoft.Extensions.Time.Testing; + +using PowerSync.Common.Client; +using PowerSync.Common.Client.Connection; +using PowerSync.Common.Client.Sync; +using PowerSync.Common.Client.Sync.Bucket; +using PowerSync.Common.Client.Sync.Stream; +using PowerSync.Common.Tests.Utils; +using PowerSync.Common.Tests.Utils.Sync; + +namespace PowerSync.Common.Tests.Client.Sync; + +/// +/// dotnet test -v n --framework net8.0 --filter "CheckpointRequestsTests" +/// +public class CheckpointRequestsTests : IAsyncLifetime +{ + MockSyncService _syncService = null!; + PowerSyncDatabase _db = null!; + + private static PowerSyncConnectionOptions WithRequests(int? retryDelayMs = null) => + new(checkpointMode: new CheckpointMode.Requests(), retryDelayMs: retryDelayMs); + + public async Task InitializeAsync() + { + _syncService = new MockSyncService(); + _db = _syncService.CreateDatabase(); + await _db.Init(); + } + + public async Task DisposeAsync() + { + await _db.Disconnect(); + await _db.Close(); + _syncService.Close(); + DatabaseUtils.CleanDb(_db.Database.Name); + } + + [Fact(Timeout = 15000)] + public async Task CheckpointRequests_WarnsCustomConnectorWithoutRequestsEnabled() + { + await _db.Connect(new CheckpointRequestConnector()); + + var logs = _syncService.Logs; + Assert.Single(logs); + Assert.Contains("implements ICustomCheckpointRequestConnector, but Connect() was called without checkpoint requests enabled.", logs[0].Message); + } + + [Fact(Timeout = 15000)] + public async Task CheckpointRequests_RequestsCheckpointsForUpdates() + { + await _db.Connect(new TestConnector(), WithRequests()); + + // Every iteration reconciles its checkpoint state with the service before requests are allowed. + await TestUtils.WaitForAsync(() => _syncService.CheckpointRequests.Count == 1); + + await _db.Execute("INSERT INTO lists (id, name) VALUES (?, ?)", ["id", "local write"]); + var watched = _db.Watch("SELECT name FROM lists", null, new() { TriggerImmediately = true }).GetAsyncEnumerator(); + await watched.MoveNextAsync(); + + Assert.Single(watched.Current); + Assert.Equal("local write", watched.Current[0].name); + + // The local write should eventually be uploaded, which requests a checkpoint. + await TestUtils.WaitForAsync(() => _syncService.CheckpointRequests.Count == 2); + + _syncService.PushLine(new StreamingSyncCheckpoint + { + Checkpoint = new() + { + LastOpId = "1", + Buckets = [MockDataFactory.Bucket("a", 1, subscriptions: Array.Empty())], + WriteCheckpoint = _syncService.LastWriteCheckpoint.ToString(), + } + }); + _syncService.PushLine(new StreamingSyncDataJSON + { + Data = new SyncDataBucketJSON + { + Bucket = "a", + Data = [ + new OplogEntryJSON + { + Checksum = 0, + OpId = "1", + ObjectId = "id", + ObjectType = "lists", + Op = "REMOVE", + } + ] + } + }); + _syncService.PushLine(new StreamingSyncCheckpointComplete { CheckpointComplete = new() { LastOpId = "1" } }); + + await watched.MoveNextAsync(); + Assert.Empty(watched.Current); + } + + [Fact(Timeout = 15000)] + public async Task CheckpointRequests_ReportsDownloadErrorWhenRequestingCheckpointFails() + { + _syncService.CheckpointRequestsSupported = false; + + // Connect() resolves once connected, which never happens here. + _ = _db.Connect(new TestConnector(), WithRequests()); + + await TestUtils.WaitForAsync(() => _db.CurrentStatus.DataFlowStatus.DownloadError != null); + + Assert.False(_db.CurrentStatus.Connected); + Assert.Contains("/sync/checkpoint-request", _db.CurrentStatus.DataFlowStatus.DownloadError!.Message); + } + + /// + /// The service is allowed to forget checkpoint requests, so an unapplied one has to be re-posted + /// until it is. Uses a fake clock to skip the (minimum 10s) retry delay, the same way the JS and + /// Kotlin equivalents of this test use their frameworks' virtual time. + /// + [Fact(Timeout = 30000)] + public async Task CheckpointRequests_RepostsCurrentCheckpointUntilApplied() + { + var time = new FakeTimeProvider(); + await using var fake = new FakeClockDatabase(_syncService, time); + + await fake.Db.Connect(new TestConnector(), WithRequests()); + + // Wait for the initial post (seed). + await AdvanceUntil(time, () => _syncService.CheckpointRequests.Count >= 1); + + await fake.Db.Execute("INSERT INTO lists (id, name) VALUES (?, ?)", ["id", "local write"]); + await AdvanceUntil(time, () => _syncService.CheckpointRequests.Count >= 2); + + var requested = _syncService.CheckpointRequests[^1]; + + // Nothing acknowledged it, so the same id keeps being posted. + for (var i = 3; i <= 6; i++) + { + await AdvanceUntil(time, () => _syncService.CheckpointRequests.Count >= i); + Assert.Equal(requested, _syncService.CheckpointRequests[^1]); + } + + // Finally, include the checkpoint. + _syncService.PushLine(new StreamingSyncCheckpoint + { + Checkpoint = new() + { + LastOpId = "0", + Buckets = [], + WriteCheckpoint = _syncService.LastWriteCheckpoint.ToString(), + } + }); + _syncService.PushLine(new StreamingSyncCheckpointComplete { CheckpointComplete = new() { LastOpId = "0" } }); + await fake.Db.WaitForFirstSync(); + + // Which means we shouldn't keep requesting it. + var totalRequests = _syncService.CheckpointRequests.Count; + for (var i = 0; i < 20; i++) + { + time.Advance(TimeSpan.FromMinutes(3)); + await Task.Yield(); + } + await Task.Delay(200); + Assert.Equal(totalRequests, _syncService.CheckpointRequests.Count); + } + + /// + /// Drives forward until holds, yielding to + /// the real scheduler in between so the sync loops can make progress. + /// + private static async Task AdvanceUntil( + FakeTimeProvider time, + Func condition, + TimeSpan? timeout = null) + { + var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(10)); + while (!condition()) + { + if (DateTime.UtcNow > deadline) + { + throw new TimeoutException("Condition not met before the (real time) timeout"); + } + + time.Advance(TimeSpan.FromSeconds(1)); + await Task.Delay(5); + } + } + + /// A database on a fake clock, torn down independently of the shared one. + private sealed class FakeClockDatabase : IAsyncDisposable + { + public PowerSyncDatabase Db { get; } + + public FakeClockDatabase(MockSyncService syncService, FakeTimeProvider time) + { + Db = syncService.CreateDatabase(timeProvider: time); + Db.Init().GetAwaiter().GetResult(); + } + + public async ValueTask DisposeAsync() + { + var name = Db.Database.Name; + await Db.Disconnect(); + await Db.Close(); + DatabaseUtils.CleanDb(name); + } + } + + /// + /// A checkpoint request needs a seeded download iteration, so wanting one has to cut a pending + /// retry delay short instead of waiting it out. + /// + [Fact(Timeout = 30000)] + public async Task CheckpointRequests_DownloadIsRetriedOnCheckpointRequest() + { + await _db.Connect(new TestConnector(), WithRequests(retryDelayMs: 10_000)); + await TestUtils.WaitForAsync(() => _syncService.CheckpointRequests.Count >= 1); + + var iterationsBefore = _syncService.Requests.Count; + + // Destroy the connection by sending a bogus line. + _syncService.PushLine(new StreamingSyncCheckpoint + { + Checkpoint = new() { LastOpId = "invalid line", Buckets = [] } + }); + await TestUtils.WaitForAsync(() => _db.CurrentStatus.DataFlowStatus.DownloadError != null); + + var start = DateTime.UtcNow; + await _db.Execute("INSERT INTO lists (id, name) VALUES (uuid(), ?)", ["restart plz"]); + + await TestUtils.WaitForAsync( + () => _syncService.Requests.Count > iterationsBefore, + TimeSpan.FromSeconds(8)); + + var elapsed = DateTime.UtcNow - start; + Assert.True( + elapsed < TimeSpan.FromSeconds(8), + $"Reconnected after {elapsed.TotalSeconds:F1}s, expected the 10s retry delay to be cut short."); + } + + [Fact(Timeout = 15000)] + public async Task CheckpointRequests_CanUseCheckpointMethodFromConnector() + { + var didRequestCheckpoint = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var connector = new TestCustomCheckpointsConnector((_, requestId, _) => + { + didRequestCheckpoint.TrySetResult(requestId); + return Task.FromResult(requestId); + }); + + await _db.Connect(connector, WithRequests()); + + Assert.Equal(1, await didRequestCheckpoint.Task); + + // The custom implementation replaces the request to the service. + Assert.Empty(_syncService.CheckpointRequests); + } + + /// + /// Simulates switching users after the old token expired: the client expects a checkpoint of 100, + /// which the service wouldn't have for another user yet. Posting the existing id lets the service + /// recognise that this device + user combination needs higher checkpoint ids. + /// + [Fact(Timeout = 15000)] + public async Task CheckpointRequests_ReconcilesCheckpointStateOnTokenExpiry() + { + _syncService.LastWriteCheckpoint = 100; + + await _db.Connect(new TestConnector(), WithRequests(retryDelayMs: 200)); + await TestUtils.WaitForAsync(() => _syncService.CheckpointRequests.Count == 1); + + _syncService.LastWriteCheckpoint = 0; + _syncService.PushLine(new StreamingSyncKeepalive { TokenExpiresIn = 0 }); + + await TestUtils.WaitForAsync( + () => _syncService.CheckpointRequests.Count >= 2, + TimeSpan.FromSeconds(10)); + Assert.Equal(100, _syncService.LastWriteCheckpoint); + } + + /// + /// Seeding runs alongside line processing rather than blocking it. + /// + [Fact(Timeout = 15000)] + public async Task CheckpointRequests_ReadsSyncLinesBeforeCheckpointRequestsAreReady() + { + var hasInitialRequest = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var completeInitialRequest = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _syncService.BeforeCheckpointRequestResponse = async () => + { + hasInitialRequest.TrySetResult(true); + await completeInitialRequest.Task; + }; + + _ = _db.Connect(new TestConnector(), WithRequests()); + await hasInitialRequest.Task; + + _syncService.PushLine(new StreamingSyncCheckpoint + { + Checkpoint = new() { LastOpId = "0", Buckets = [], WriteCheckpoint = "1" } + }); + + await TestUtils.WaitForAsync(() => _db.CurrentStatus.DataFlowStatus.Downloading); + completeInitialRequest.TrySetResult(true); + } + + [Fact(Timeout = 15000)] + public async Task CheckpointRequests_FailsWhenDisconnected() + { + var exception = await Assert.ThrowsAsync(() => _db.RequestCheckpoint()); + Assert.Equal(CheckpointRequestException.Disconnected, exception.Message); + } + + [Fact(Timeout = 5000)] + public async Task CheckpointRequests_FailsWhenConnectedInLegacyMode() + { + await _db.Connect(new TestConnector(), new() { CheckpointMode = CheckpointMode.Legacy }); + + var exception = await Assert.ThrowsAsync(() => _db.RequestCheckpoint()); + Assert.Equal(CheckpointRequestException.Disabled, exception.Message); + } + + [Fact(Timeout = 5000)] + public async Task CheckpointRequests_WaitsUntilDataIsApplied() + { + await _db.Connect(new TestConnector(), new() { CheckpointMode = new CheckpointMode.Requests() }); + + var checkpoint = await _db.RequestCheckpoint(); + _syncService.PushLine(new StreamingSyncCheckpoint + { + Checkpoint = new Checkpoint { LastOpId = "0", WriteCheckpoint = "2", }, + }); + Assert.False(checkpoint.HasSynced); + _syncService.PushLine(MockDataFactory.CheckpointComplete("0")); + + await checkpoint.WaitForSync(); + Assert.True(checkpoint.HasSynced); + } + + [Fact(Timeout = 5000)] + public async Task CheckpointRequests_ThrowsOnDisconnectButCanConnectAgain() + { + await _db.Connect(new TestConnector(), new() { CheckpointMode = new CheckpointMode.Requests() }); + var checkpoint = await _db.RequestCheckpoint(); + + var didThrowCorrectly = false; + var waitForSyncTask = Task.Run(async () => + { + try + { + await checkpoint.WaitForSync(); + // Expected WaitForSync to throw + didThrowCorrectly = false; + } + catch (CheckpointRequestException ex) + { + didThrowCorrectly = ex.Message == CheckpointRequestException.Disconnected; + } + catch + { + didThrowCorrectly = false; + } + }); + + await _db.Disconnect(); + await waitForSyncTask; + Assert.True(didThrowCorrectly); + + await _db.Connect(new TestConnector(), new() { CheckpointMode = new CheckpointMode.Requests() }); + _syncService.PushLine(new StreamingSyncCheckpoint + { + Checkpoint = new Checkpoint { LastOpId = "0", WriteCheckpoint = "2", }, + }); + _syncService.PushLine(MockDataFactory.CheckpointComplete("0")); + await checkpoint.WaitForSync(); + } + + [Fact(Timeout = 5000)] + public async Task CheckpointRequests_FailsWhenReconnectingInLegacyMode() + { + await _db.Connect(new TestConnector(), new() { CheckpointMode = new CheckpointMode.Requests() }); + var checkpoint = await _db.RequestCheckpoint(); + + await _db.Disconnect(); + await _db.Connect(new TestConnector(), new() { CheckpointMode = CheckpointMode.Legacy }); + + var exception = await Assert.ThrowsAsync(() => checkpoint.WaitForSync()); + Assert.Equal(CheckpointRequestException.Disabled, exception.Message); + } + + [Fact(Timeout = 5000)] + public async Task CheckpointRequests_FailsOnSyncErrors() + { + await _db.Connect(new TestConnector(), new() { CheckpointMode = new CheckpointMode.Requests() }); + var checkpoint = await _db.RequestCheckpoint(); + + var didThrowCorrectly = false; + var waitForSyncTask = Task.Run(async () => + { + try + { + await checkpoint.WaitForSync(); + // Expected WaitForSync to throw + didThrowCorrectly = false; + } + catch (CheckpointRequestException ex) + { + didThrowCorrectly = ex.Message.Contains(CheckpointRequestException.StatusError); + } + catch + { + didThrowCorrectly = false; + } + }); + + _syncService.PushLine("not a valid sync line"); + await waitForSyncTask; + Assert.True(didThrowCorrectly); + } + + [Fact(Timeout = 5000)] + public async Task CheckpointRequests_CanAbortCustomCheckpointRequest() + { + var connector = new StagedCheckpointRequestConnector(); + await _db.Connect(connector, WithRequests()); + await TestUtils.WaitForAsync(() => connector.Launched); + + // Simulate the database disconnecting mid-request. + var disconnectTask = _db.Disconnect(); + connector.Continue(); + await disconnectTask; + + await TestUtils.WaitForAsync(() => connector.Canceled); + Assert.False(connector.Completed); + } + + [Fact(Timeout = 5000)] + public async Task CheckpointRequests_RequestThrowsIfCanceledImmediately() + { + var canceledCts = new CancellationTokenSource(); + canceledCts.Cancel(); + + await _db.Connect(new TestConnector(), WithRequests()); + + await Assert.ThrowsAsync(() => _db.RequestCheckpoint(canceledCts.Token)); + } + + [Fact(Timeout = 5000)] + public async Task CheckpointRequests_WaitForSyncThrowsIfCanceledImmediately() + { + var canceledCts = new CancellationTokenSource(); + canceledCts.Cancel(); + + await _db.Connect(new TestConnector(), WithRequests()); + var checkpoint = await _db.RequestCheckpoint(); + + await Assert.ThrowsAsync(() => checkpoint.WaitForSync(canceledCts.Token)); + } + + /// + /// Cancellation has to reach the in-flight request itself, not just guard the entry point. + /// + [Fact(Timeout = 15000)] + public async Task CheckpointRequests_RequestThrowsIfCanceledWhileInFlight() + { + var blocked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var calls = 0; + + // The first request is the seed posted by the download iteration, which has to complete + // before explicit requests are allowed. Only the request under test hangs, so that the + // retry afterwards can resolve. + var connector = new TestCustomCheckpointsConnector(async (_, requestId, token) => + { + if (Interlocked.Increment(ref calls) == 2) + { + blocked.TrySetResult(); + await Task.Delay(Timeout.Infinite, token); + } + + return requestId; + }); + + await _db.Connect(connector, WithRequests()); + + using var cts = new CancellationTokenSource(); + var request = _db.RequestCheckpoint(cts.Token); + await blocked.Task; + + cts.Cancel(); + await Assert.ThrowsAnyAsync(() => request); + + // The exclusive lock has to be released again, otherwise later requests would deadlock. + Assert.Equal(2, calls); + var checkpoint = await _db.RequestCheckpoint(); + Assert.False(checkpoint.HasSynced); + Assert.Equal(3, calls); + } + + /// + /// Requests park until a download iteration has reconciled checkpoint state with the service, + /// which is another point at which the caller can give up. + /// + [Fact(Timeout = 15000)] + public async Task CheckpointRequests_RequestThrowsIfCanceledWhileWaitingForSeed() + { + // A retry delay long enough that only a parked request can wake the download loop. + await _db.Connect(new TestConnector(), WithRequests(retryDelayMs: 60_000)); + await TestUtils.WaitForAsync(() => _syncService.CheckpointRequests.Count >= 1); + + // Leave the next iteration's seed unanswered, so checkpoint state stays pending. + var seedStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var completeSeed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _syncService.BeforeCheckpointRequestResponse = async () => + { + seedStarted.TrySetResult(true); + await completeSeed.Task; + }; + + // Destroy the connection with a bogus line: checkpoint requests are no longer ready. + _syncService.PushLine(new StreamingSyncCheckpoint + { + Checkpoint = new() { LastOpId = "invalid line", Buckets = [] } + }); + await TestUtils.WaitForAsync(() => _db.CurrentStatus.DataFlowStatus.DownloadError != null); + + using var cts = new CancellationTokenSource(); + var request = _db.RequestCheckpoint(cts.Token); + + // Parking cuts the retry delay short, and the restarted iteration's seed is the one being + // held up above, so the request is still waiting on it here. + await seedStarted.Task; + Assert.False(request.IsCompleted); + + cts.Cancel(); + await Assert.ThrowsAnyAsync(() => request); + + completeSeed.TrySetResult(true); + } + + [Fact(Timeout = 15000)] + public async Task CheckpointRequests_WaitForSyncThrowsIfCanceledWhileWaiting() + { + await _db.Connect(new TestConnector(), WithRequests()); + var checkpoint = await _db.RequestCheckpoint(); + + var listenersBefore = _db.Events.OnStatusChanged.SubscriberCount(); + + using var cts = new CancellationTokenSource(); + var wait = checkpoint.WaitForSync(cts.Token); + + // Nothing has been applied, so the wait is parked on sync status updates. + await TestUtils.WaitForAsync(() => _db.Events.OnStatusChanged.SubscriberCount() > listenersBefore); + Assert.False(wait.IsCompleted); + + cts.Cancel(); + await Assert.ThrowsAnyAsync(() => wait); + Assert.False(checkpoint.HasSynced); + + // Abandoning a wait doesn't invalidate the request, it can still be awaited again. + _syncService.PushLine(new StreamingSyncCheckpoint + { + Checkpoint = new Checkpoint { LastOpId = "0", WriteCheckpoint = "2", }, + }); + _syncService.PushLine(MockDataFactory.CheckpointComplete("0")); + + await checkpoint.WaitForSync(); + Assert.True(checkpoint.HasSynced); + } + + /// + /// Cancelling the wait must not take the checkpoint's other waiters down with it. + /// + [Fact(Timeout = 15000)] + public async Task CheckpointRequests_WaitForSyncCancellationIsPerCaller() + { + await _db.Connect(new TestConnector(), WithRequests()); + var checkpoint = await _db.RequestCheckpoint(); + + var listenersBefore = _db.Events.OnStatusChanged.SubscriberCount(); + + using var cts = new CancellationTokenSource(); + var canceledWait = checkpoint.WaitForSync(cts.Token); + var survivingWait = checkpoint.WaitForSync(); + + // Both waits listen for status updates; neither can be resolved before one is applied. + await TestUtils.WaitForAsync(() => _db.Events.OnStatusChanged.SubscriberCount() >= listenersBefore + 2); + + cts.Cancel(); + await Assert.ThrowsAnyAsync(() => canceledWait); + Assert.False(survivingWait.IsCompleted); + + _syncService.PushLine(new StreamingSyncCheckpoint + { + Checkpoint = new Checkpoint { LastOpId = "0", WriteCheckpoint = "2", }, + }); + _syncService.PushLine(MockDataFactory.CheckpointComplete("0")); + + await survivingWait; + Assert.True(checkpoint.HasSynced); + } + + // A class with a settable property rather than a positional record: Dapper can't pick a + // constructor when the result set is empty and SQLite reports no column type. + private class NameResult + { + public string name { get; set; } = ""; + } +} + +class CheckpointRequestConnector : TestConnector, ICustomCheckpointRequestConnector +{ + public Task PostCheckpointRequest(string clientId, long requestId, CancellationToken token) + { + return Task.FromResult(requestId); + } +} + +// Runs a single, multi-stage PostCheckpointRequest call. +class StagedCheckpointRequestConnector(bool enableConsole = false) : TestConnector, ICustomCheckpointRequestConnector +{ + private readonly bool _enableConsole = enableConsole; + + private readonly TaskCompletionSource _continueTcs = new(); + + private bool _launched = false; + private bool _completed = false; + private bool _canceled = false; + + public bool Launched + { + get { lock (_lock) { return _launched; } } + private set { lock (_lock) { _launched = value; } } + } + public bool Completed + { + get { lock (_lock) { return _completed; } } + private set { lock (_lock) { _completed = value; } } + } + public bool Canceled + { + get { lock (_lock) { return _canceled; } } + private set { lock (_lock) { _canceled = value; } } + } + + private readonly object _lock = new(); + + public void Continue() + { + _continueTcs.TrySetResult(); + } + + private void ThrowIfCancellationRequested(CancellationToken ct) + { + if (ct.IsCancellationRequested) + { + Debug("Cancellation requested, throwing OperationCanceledException."); + Canceled = true; + throw new OperationCanceledException(); + } + } + + // Used to debug and diagnose a malformed test. + private void Debug(string message) + { + if (_enableConsole) + Console.WriteLine($"[CheckpointRequestConnector] {message}"); + } + + public async Task PostCheckpointRequest(string clientId, long requestId, CancellationToken token) + { + Launched = true; + Debug("Launched."); + + ThrowIfCancellationRequested(token); + Debug("First guard cleared, awaiting continue..."); + + await _continueTcs.Task; + Debug("Continue received."); + ThrowIfCancellationRequested(token); + Debug("Second guard cleared."); + + Completed = true; + Debug("Completed."); + + return requestId; + } +} diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs index 69294c33..1dd5b436 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs @@ -9,7 +9,7 @@ namespace PowerSync.Common.Tests.Client.Sync; using PowerSync.Common.Tests.Utils.Sync; /// -/// dotnet test -v n --framework net8.0 --filter "StreamingSyncRetryTests" +/// dotnet test -v n --framework net10.0 --filter "StreamingSyncRetryTests" /// public class StreamingSyncRetryTests { @@ -96,10 +96,10 @@ SemaphoreSlim signal ); } - public override Task Get(string path, Dictionary? headers = null) + public override Task FetchJson(string path, HttpMethod? method = null, object? data = null, Dictionary? headers = null, CancellationToken ct = default) { - var response = new StreamingSyncImplementation.ApiResponse( - new StreamingSyncImplementation.ResponseData(1) + var response = new StreamingSyncImplementation.LegacyWriteCheckpointApiResponse( + new StreamingSyncImplementation.LegacyWriteCheckpointResponseData(1) ); return Task.FromResult((T)(object)response); } diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationControlFlowTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationControlFlowTests.cs index 43c86914..fe94e53b 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationControlFlowTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationControlFlowTests.cs @@ -201,6 +201,54 @@ public async Task HideDisconnectRequestsImmediateRestart() "Expected CloseSyncStream(hide_disconnect: true) to request an immediate restart."); } + /// + /// A CRUD upload pass that finishes while no download iteration is running must still reach the + /// core extension. The core holds a checkpoint back while it believes local writes are still + /// pending, so a dropped notification leaves downloaded data unapplied until the next local + /// write happens to produce another notification. + /// Surfaces on connect (the upload loop starts alongside the download loop) and between retries. + /// + [Fact(Timeout = 15000)] + public async Task UploadCompletedWithNoActiveIterationStillReachesCore() + { + var uploadPassReachedEnd = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var completedUpload = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + var adapter = new ScriptedAdapter( + (op, _) => + { + switch (op) + { + case PowerSyncControlCommand.START: + return EstablishOnly; + case PowerSyncControlCommand.NOTIFY_CRUD_UPLOAD_COMPLETED: + completedUpload.TrySetResult(true); + return NoInstructions; + default: + return NoInstructions; + } + }, + onUpdateLocalTarget: () => uploadPassReachedEnd.TrySetResult(true)); + + // Stream stays open so the control loop keeps consuming. + var harness = Harness.Create(adapter, _ => Task.FromResult(new HangingStream(""))); + + // The upload pass runs and finishes before any iteration exists to be notified. + var crudLoop = harness.RunCrudUploadLoop(); + await uploadPassReachedEnd.Task; + + var iteration = harness.RunIteration(); + var forwarded = await Task.WhenAny(completedUpload.Task, Task.Delay(5000)) == completedUpload.Task; + + harness.Cancel(); + try { await iteration; } catch { /* teardown */ } + try { await crudLoop; } catch { /* teardown */ } + + Assert.True(forwarded, + "Expected 'completed_upload' to reach the core from the iteration that started after the " + + $"upload completed, but only saw: {string.Join(", ", adapter.Ops)}"); + } + // ---- harness ----------------------------------------------------------- private sealed class Harness @@ -223,6 +271,8 @@ public static Harness Create(ScriptedAdapter adapter, Func RunIteration() => sync.RunIteration(cts.Token); + public Task RunCrudUploadLoop() => sync.RunCrudUploadLoop(cts.Token); + public void Cancel() => cts.Cancel(); } @@ -235,10 +285,15 @@ private sealed class TestSyncImplementation(StreamingSyncImplementationOptions o var result = await RustStreamingSyncIteration(token, DEFAULT_STREAM_CONNECTION_OPTIONS); return result.ImmediateRestart; } + + public Task RunCrudUploadLoop(CancellationToken token) => + CrudUploadLoop(token, new PowerSyncConnectionOptions(crudUploadThrottleMs: 0)); } /// Records every powersync_control op and replies with canned instructions. - private sealed class ScriptedAdapter(Func respond) : IBucketStorageAdapter + private sealed class ScriptedAdapter( + Func respond, + Action? onUpdateLocalTarget = null) : IBucketStorageAdapter { private readonly ConcurrentQueue ops = new(); @@ -255,8 +310,15 @@ public Task Control(string op, object? payload) public Task NextCrudItem() => Task.FromResult(null); public Task HasCrud() => Task.FromResult(false); public Task GetCrudBatch(int limit = 100) => Task.FromResult(null); - public Task UpdateLocalTarget(Func> callback) => Task.FromResult(false); + + public Task UpdateLocalTarget(Func> callback) + { + onUpdateLocalTarget?.Invoke(); + return Task.FromResult(false); + } + public Task HandleCrudCheckpoint(long lastClientId, long? writeCheckpoint = null) => Task.CompletedTask; + public Task ReadOrUpdateCheckpoint(string variant, long? update = null) => Task.FromResult(1); public Task GetClientId() => Task.FromResult("test-client"); public void Close() { } } diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs index 88a4972f..6ee91e9b 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs @@ -7,7 +7,7 @@ namespace PowerSync.Common.Tests.Client.Sync; using PowerSync.Common.Tests.Utils.Sync; /// -/// dotnet test -v n --framework net8.0 --filter "SyncStreamsTests" +/// dotnet test -v n --framework net10.0 --filter "SyncStreamsTests" /// public class SyncStreamsTests : IAsyncLifetime { @@ -120,6 +120,9 @@ public async Task SubscribesWithStreams() syncService.PushLine(MockDataFactory.CheckpointComplete(lastOpId: "0")); await a.WaitForFirstSync(); + + a.Unsubscribe(); + b.Unsubscribe(); } [Fact] @@ -176,6 +179,7 @@ public async Task SubscriptionsUpdateWhileOfflineTest() var status = await statusTask; Assert.NotNull(status.ForStream(subscription)); + subscription.Unsubscribe(); } // FIx diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncTests.cs index a10f3b4a..09c5effa 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncTests.cs @@ -6,7 +6,7 @@ namespace PowerSync.Common.Tests.Client.Sync; /// -/// dotnet test -v n --framework net8.0 --filter "SyncTests" +/// dotnet test -v n --framework net10.0 --filter "SyncTests" /// public class SyncTests : IAsyncLifetime { diff --git a/Tests/PowerSync/PowerSync.Common.Tests/DB/SchemaTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/DB/SchemaTests.cs index 20c6e290..29aea943 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/DB/SchemaTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/DB/SchemaTests.cs @@ -7,7 +7,7 @@ namespace PowerSync.Common.Tests.DB.Schema; using PowerSync.Common.Tests; /// -/// dotnet test -v n --framework net8.0 --filter "SchemaTests" +/// dotnet test -v n --framework net10.0 --filter "SchemaTests" /// public class SchemaTests { @@ -201,7 +201,7 @@ public void AttributeParser_Logs_Test() class Invalid1 { public string id { get; set; } = ""; } [Fact] - public async void AttributeParser_InvalidSchema_1() + public async Task AttributeParser_InvalidSchema_1() { var ex = await Assert.ThrowsAsync(() => { @@ -214,7 +214,7 @@ public async void AttributeParser_InvalidSchema_1() [Table("invalid")] class Invalid2 { } [Fact] - public async void AttributeParser_InvalidSchema_2() + public async Task AttributeParser_InvalidSchema_2() { var ex = await Assert.ThrowsAsync(() => { @@ -227,7 +227,7 @@ public async void AttributeParser_InvalidSchema_2() [Table("invalid")] class Invalid3 { public int id { get; set; } } [Fact] - public async void AttributeParser_InvalidSchema_3() + public async Task AttributeParser_InvalidSchema_3() { var ex = await Assert.ThrowsAsync(() => { @@ -244,7 +244,7 @@ class Invalid4 public string id { get; set; } = ""; } [Fact] - public async void AttributeParser_InvalidSchema_4() + public async Task AttributeParser_InvalidSchema_4() { var ex = await Assert.ThrowsAsync(() => { @@ -261,7 +261,7 @@ class Invalid5 public Invalid1 invalid_type { get; set; } = default!; } [Fact] - public async void AttributeParser_InvalidSchema_5() + public async Task AttributeParser_InvalidSchema_5() { var ex = await Assert.ThrowsAsync(() => { @@ -277,7 +277,7 @@ class Invalid6 public string id { get; set; } = ""; } [Fact] - public async void AttributeParser_InvalidSchema_6() + public async Task AttributeParser_InvalidSchema_6() { var ex = await Assert.ThrowsAsync(() => { @@ -293,7 +293,7 @@ class Invalid7 public string id { get; set; } = ""; } [Fact] - public async void AttributeParser_InvalidSchema_7() + public async Task AttributeParser_InvalidSchema_7() { var ex = await Assert.ThrowsAsync(() => { diff --git a/Tests/PowerSync/PowerSync.Common.Tests/GlobalUsing.cs b/Tests/PowerSync/PowerSync.Common.Tests/GlobalUsing.cs deleted file mode 100644 index cca77378..00000000 --- a/Tests/PowerSync/PowerSync.Common.Tests/GlobalUsing.cs +++ /dev/null @@ -1 +0,0 @@ -global using FactAttribute = PowerSync.Common.Tests.Utils.FactAttribute; diff --git a/Tests/PowerSync/PowerSync.Common.Tests/MDSQLite/MDSQLiteAdapterTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/MDSQLite/MDSQLiteAdapterTests.cs index d78e262a..41843828 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/MDSQLite/MDSQLiteAdapterTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/MDSQLite/MDSQLiteAdapterTests.cs @@ -8,7 +8,7 @@ namespace PowerSync.Common.Tests.MDSQLite; using PowerSync.Common.Utils; /// -/// dotnet test -v n --framework net8.0 --filter "MDSQLiteAdapterTests" +/// dotnet test -v n --framework net10.0 --filter "MDSQLiteAdapterTests" /// [Collection("MDSQLiteAdapterTests")] public class MDSQLiteAdapterTests diff --git a/Tests/PowerSync/PowerSync.Common.Tests/PowerSync.Common.Tests.csproj b/Tests/PowerSync/PowerSync.Common.Tests/PowerSync.Common.Tests.csproj index a153f06b..ca009fc9 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/PowerSync.Common.Tests.csproj +++ b/Tests/PowerSync/PowerSync.Common.Tests/PowerSync.Common.Tests.csproj @@ -1,7 +1,7 @@ - net6.0;net8.0;net9.0 + net8.0;net9.0;net10.0 12 enable enable @@ -11,14 +11,15 @@ - - - - - - - - + + runtime; build; native; contentfiles; analyzers; buildtransitive + all + + + + + + runtime; build; native; contentfiles; analyzers; buildtransitive all diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Utils/FactAttribute.cs b/Tests/PowerSync/PowerSync.Common.Tests/Utils/FactAttribute.cs deleted file mode 100644 index 53c67cfd..00000000 --- a/Tests/PowerSync/PowerSync.Common.Tests/Utils/FactAttribute.cs +++ /dev/null @@ -1,10 +0,0 @@ -namespace PowerSync.Common.Tests.Utils; - -[AttributeUsage(AttributeTargets.Method)] -public class FactAttribute : Xunit.FactAttribute -{ - public FactAttribute() - { - Timeout = 5000; - } -} diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs b/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs index e37baed0..54c9b80e 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs @@ -1,4 +1,4 @@ -using System.Dynamic; +using System.Collections.Concurrent; using System.IO.Pipelines; using System.Text; @@ -19,10 +19,66 @@ namespace PowerSync.Common.Tests.Utils.Sync; public class MockSyncService : EventStream { - private readonly List _requests = new(); - + private readonly List _requests = []; public IReadOnlyList Requests => _requests; + private readonly ListLoggerProvider _listLoggerProvider = new(); + public IReadOnlyList Logs => _listLoggerProvider.Logs; + + private readonly object checkpointGate = new(); + private readonly List checkpointRequests = []; + private long lastWriteCheckpoint; + + /// + /// The highest checkpoint request id this service has handed out. Settable so tests can simulate a + /// client whose local counter has drifted from the service's. + /// + public long LastWriteCheckpoint + { + get { lock (checkpointGate) { return lastWriteCheckpoint; } } + set { lock (checkpointGate) { lastWriteCheckpoint = value; } } + } + + /// Every checkpoint request id received on `/sync/checkpoint-request`, in order. + public IReadOnlyList CheckpointRequests + { + get { lock (checkpointGate) { return [.. checkpointRequests]; } } + } + + /// Set to false to emulate a service too old to know `/sync/checkpoint-request`. + public bool CheckpointRequestsSupported { get; set; } = true; + + /// Runs after a checkpoint request is recorded, but before it is answered. + public Func BeforeCheckpointRequestResponse { get; set; } = () => Task.CompletedTask; + + /// + /// Answers a checkpoint request the way the service does: the effective id is the higher of the + /// requested id and the one the service already knows about. + /// + internal async Task HandleCheckpointRequest(CheckpointRequestPayload request) + { + if (!CheckpointRequestsSupported) + { + throw new HttpRequestException( + "Received NotFound - Not Found when getting from /sync/checkpoint-request: "); + } + + long resolved; + lock (checkpointGate) + { + checkpointRequests.Add(request.CheckpointRequestId); + resolved = Math.Max(lastWriteCheckpoint, request.CheckpointRequestId); + lastWriteCheckpoint = resolved; + } + + await BeforeCheckpointRequestResponse(); + + return new CheckpointRequestResponse + { + Data = new CheckpointRequestResponseData { CheckpointRequestId = resolved } + }; + } + public void PushLine(StreamingSyncLine line) { Emit(JsonConvert.SerializeObject(line)); @@ -33,7 +89,7 @@ public void PushLine(string line) Emit(line); } - public PowerSyncDatabase CreateDatabase(string? dbFilename = null) + public PowerSyncDatabase CreateDatabase(string? dbFilename = null, TimeProvider? timeProvider = null) { dbFilename ??= $"sync-stream-{Guid.NewGuid():N}.db"; var connector = new TestConnector(); @@ -44,16 +100,18 @@ public PowerSyncDatabase CreateDatabase(string? dbFilename = null) Database = new SQLOpenOptions { DbFilename = dbFilename }, Schema = TestSchemaTodoList.AppSchema, RemoteFactory = _ => mockRemote, - Logger = createLogger() + TimeProvider = timeProvider, + Logger = CreateLogger() }); } - private ILogger createLogger() + private ILogger CreateLogger() { ILoggerFactory loggerFactory = LoggerFactory.Create(builder => { builder.AddConsole(); - builder.SetMinimumLevel(LogLevel.Error); + builder.AddProvider(_listLoggerProvider); + builder.SetMinimumLevel(LogLevel.Warning); }); return loggerFactory.CreateLogger("PowerSyncLogger"); } @@ -156,41 +214,55 @@ public MockRemote( public override Task PostStreamRaw(SyncStreamOptions options) { - connectedListeners.Add(options.Data); + if (options.Path.EndsWith("/sync/stream")) + { + connectedListeners.Add(options.Data); - var pipe = new Pipe(); - var writer = pipe.Writer; + var pipe = new Pipe(); + var writer = pipe.Writer; - var cts = CancellationTokenSource.CreateLinkedTokenSource(options.CancellationToken); - var listener = syncService.ListenAsync(cts.Token); - _ = Task.Run(async () => - { - try + var cts = CancellationTokenSource.CreateLinkedTokenSource(options.CancellationToken); + var listener = syncService.ListenAsync(cts.Token); + _ = Task.Run(async () => { - await foreach (var line in listener) + try { - var bytes = Encoding.UTF8.GetBytes(line + "\n"); - await writer.WriteAsync(bytes); + await foreach (var line in listener) + { + var bytes = Encoding.UTF8.GetBytes(line + "\n"); + await writer.WriteAsync(bytes); + } } - } - finally - { - await writer.CompleteAsync(); - cts.Cancel(); - cts.Dispose(); - } - }); + finally + { + await writer.CompleteAsync(); + cts.Cancel(); + cts.Dispose(); + } + }); + + return Task.FromResult(pipe.Reader.AsStream()); + } - return Task.FromResult(pipe.Reader.AsStream()); + throw new InvalidOperationException($"MockRemote received an unexpected stream request: {options.Path}"); } - public override Task Get(string path, Dictionary? headers = null) + public override async Task FetchJson(string path, HttpMethod? method = null, object? data = null, Dictionary? headers = null, CancellationToken ct = default) { - var response = new StreamingSyncImplementation.ApiResponse( - new StreamingSyncImplementation.ResponseData(1) - ); + if (path.Contains("/sync/checkpoint-request")) + { + var response = await syncService.HandleCheckpointRequest((CheckpointRequestPayload)data!); + return (T)(object)response; + } - return Task.FromResult((T)(object)response); + if (path.Contains("write-checkpoint2.json")) + { + return (T)(object)new StreamingSyncImplementation.LegacyWriteCheckpointApiResponse( + new StreamingSyncImplementation.LegacyWriteCheckpointResponseData(1) + ); + } + + throw new InvalidOperationException($"MockRemote received an unexpected request: {path}"); } } @@ -213,3 +285,40 @@ public async Task UploadData(IPowerSyncDatabase database) } } } + +public class TestCustomCheckpointsConnector(Func> postCheckpointRequest) : TestConnector, ICustomCheckpointRequestConnector +{ + private readonly Func> _postCheckpointRequest = postCheckpointRequest; + + public Task PostCheckpointRequest(string clientId, long requestId, CancellationToken ct) + => _postCheckpointRequest(clientId, requestId, ct); +} + +public record LogRecord(LogLevel LogLevel, string CategoryName, string Message, Exception? Exception); + +public class ListLogger(string categoryName, ConcurrentQueue drain) : ILogger +{ + private readonly string _categoryName = categoryName; + private readonly ConcurrentQueue _drain = drain; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + { + _drain.Enqueue(new(logLevel, _categoryName, formatter(state, exception), exception)); + } + + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; +} + +public class ListLoggerProvider : ILoggerProvider +{ + private readonly ConcurrentQueue _logs = new(); + public IReadOnlyList Logs => [.. _logs]; + + public ILogger CreateLogger(string categoryName) + { + return new ListLogger(categoryName, _logs); + } + + public void Dispose() => GC.SuppressFinalize(this); +} diff --git a/Tools/Setup/Setup.cs b/Tools/Setup/Setup.cs index c91d2ecd..e1f68ebe 100644 --- a/Tools/Setup/Setup.cs +++ b/Tools/Setup/Setup.cs @@ -10,7 +10,7 @@ /// public class PowerSyncSetup { - private const string VERSION = "0.5.2"; + private const string VERSION = "0.5.3"; private const string GITHUB_BASE_URL = $"https://github.com/powersync-ja/powersync-sqlite-core/releases/download/v{VERSION}"; @@ -113,7 +113,7 @@ public async Task SetupMauiAndroid() Directory.CreateDirectory(nativeDir); await Task.WhenAll( - DownloadAndroidLibrary("libpowersync_aarch64.android.so ", nativeDir,"arm64-v8a"), + DownloadAndroidLibrary("libpowersync_aarch64.android.so ", nativeDir, "arm64-v8a"), DownloadAndroidLibrary("libpowersync_armv7.android.so ", nativeDir, "armeabi-v7a"), DownloadAndroidLibrary("libpowersync_x86.android.so ", nativeDir, "x86"), DownloadAndroidLibrary("libpowersync_x64.android.so ", nativeDir, "x86_64") @@ -130,7 +130,7 @@ await Task.WhenAll( private async Task DownloadAndroidLibrary(string filename, string jniLibsDir, string arch) { var targetDir = Path.Combine(jniLibsDir, arch); - Directory.CreateDirectory(targetDir); + Directory.CreateDirectory(targetDir); var targetFile = Path.Combine(targetDir, "libpowersync.so"); await DownloadFile($"{GITHUB_BASE_URL}/{filename}", targetFile); } diff --git a/Tools/Setup/Setup.csproj b/Tools/Setup/Setup.csproj index 83831b3f..c5c3bf8e 100644 --- a/Tools/Setup/Setup.csproj +++ b/Tools/Setup/Setup.csproj @@ -2,6 +2,6 @@ Exe - net8.0 + net10.0 - \ No newline at end of file + diff --git a/demos/CommandLine/CommandLine.csproj b/demos/CommandLine/CommandLine.csproj index c4bc4de6..36ac2f4e 100644 --- a/demos/CommandLine/CommandLine.csproj +++ b/demos/CommandLine/CommandLine.csproj @@ -3,7 +3,7 @@ Exe 0.0.1 - net8.0;net9.0 + net8.0;net9.0;net10.0 12 enable enable @@ -12,13 +12,12 @@ - - - + + - + diff --git a/demos/MAUITodo/Data/PowerSyncData.cs b/demos/MAUITodo/Data/PowerSyncData.cs index 676781f8..d5e2d600 100644 --- a/demos/MAUITodo/Data/PowerSyncData.cs +++ b/demos/MAUITodo/Data/PowerSyncData.cs @@ -7,6 +7,8 @@ using PowerSync.Common.Attachments; using PowerSync.Common.Client; +using PowerSync.Common.Client.Sync; +using PowerSync.Common.Client.Sync.Stream; using PowerSync.Common.MDSQLite; using PowerSync.Maui.SQLite; @@ -43,7 +45,10 @@ public PowerSyncData() var nodeConnector = new NodeConnector(); UserId = nodeConnector.UserId; - Db.Connect(nodeConnector); + // Checkpoint requests let the app ask the service for a checkpoint on demand, which is what + // the pull-to-refresh gestures use. Requires PowerSync service 1.24.0 or later. + Db.Connect(nodeConnector, new PowerSyncConnectionOptions( + checkpointMode: new CheckpointMode.Requests())); var attachmentsDir = Path.Combine(FileSystem.AppDataDirectory, "attachments"); var localStorage = new FileManagerLocalStorage(attachmentsDir); @@ -76,6 +81,20 @@ private static async IAsyncEnumerable WatchTodoPhotos( private record PhotoIdResult(string photo_id); + /// + /// Asks the service for a checkpoint and waits until the local database has applied everything + /// up to it, so the caller knows the local view has caught up. + /// + /// + /// Thrown when the client is disconnected, was connected without checkpoint requests enabled, + /// or a sync error occurs before the checkpoint is applied. + /// + public async Task RefreshAsync() + { + var checkpoint = await Db.RequestCheckpoint(); + await checkpoint.WaitForSync(); + } + public async Task SaveListAsync(TodoList list) { if (list.ID != "") diff --git a/demos/MAUITodo/MAUITodo.csproj b/demos/MAUITodo/MAUITodo.csproj index 56d224b4..9f38150b 100644 --- a/demos/MAUITodo/MAUITodo.csproj +++ b/demos/MAUITodo/MAUITodo.csproj @@ -3,9 +3,9 @@ com.companyname.todo - net8.0-android;net9.0-android - $(TargetFrameworks);net8.0-windows10.0.19041.0 - $(TargetFrameworks);net8.0-ios;net8.0-maccatalyst;net9.0-ios;net9.0-maccatalyst + net9.0-android;net10.0-android + $(TargetFrameworks);net9.0-windows;net10.0-windows + $(TargetFrameworks);net9.0-ios;net9.0-maccatalyst;net10.0-ios;net10.0-maccatalyst Exe MAUITodo @@ -29,7 +29,7 @@ true true - 8.0.90 + 8.0.90 15.0 15.0 @@ -38,12 +38,6 @@ 10.0.19041.0 - - - iossimulator-x64 - true - - @@ -63,15 +57,13 @@ - - - - + + diff --git a/demos/MAUITodo/README.md b/demos/MAUITodo/README.md index ca474d8b..3df4ba1f 100644 --- a/demos/MAUITodo/README.md +++ b/demos/MAUITodo/README.md @@ -8,6 +8,17 @@ To run this demo, you need to have one of our Node.js self-host demos ([Postgres Changes made to the backend's source DB or to the self-hosted web UI will be synced to this client (and vice versa). +## Pull-to-refresh with explicit checkpoints + +The lists and todos screens support pull-to-refresh. Swiping down asks the PowerSync service for a +checkpoint via `PowerSyncDatabase.RequestCheckpoint()` and waits for the local database to apply +everything up to it with `CheckpointRequest.WaitForSync()`, so the spinner only stops once the local +view has actually caught up to the service. + +This requires connecting with `CheckpointMode.Requests()` (see `Data/PowerSyncData.cs`) and +**PowerSync service version 1.24.0 or later**. Against an older service the refresh will report a +sync error instead of completing. Checkpoint requests are currently an alpha API. + In the repo root, run the following to download the PowerSync extension: ```bash @@ -27,29 +38,29 @@ dotnet restore ### iOS ```sh -dotnet build -t:Run -f:net8.0-ios +dotnet build -t:Run -f:net10.0-ios ``` Specifyng an iOS simulator ```sh -dotnet build -t:Run -f:net8.0-ios -p:_DeviceName=:v2:udid=B1CA156A-56FC-4C3C-B35D-4BC349111FDF +dotnet build -t:Run -f:net10.0-ios -p:_DeviceName=:v2:udid=B1CA156A-56FC-4C3C-B35D-4BC349111FDF ``` ### Android ```sh -dotnet build -t:Run -f:net8.0-android +dotnet build -t:Run -f:net10.0-android ``` Specifying an Android emulator ```sh -dotnet build -t:Run -f:net8.0-android -p:_DeviceName=emulator-5554 +dotnet build -t:Run -f:net10.0-android -p:_DeviceName=emulator-5554 ``` ### MacCatalyst ```sh -dotnet build -t:Run -f:net8.0-maccatalyst +dotnet build -t:Run -f:net10.0-maccatalyst ``` diff --git a/demos/MAUITodo/Views/ListsPage.xaml b/demos/MAUITodo/Views/ListsPage.xaml index 9622ec44..6a3b1067 100644 --- a/demos/MAUITodo/Views/ListsPage.xaml +++ b/demos/MAUITodo/Views/ListsPage.xaml @@ -10,35 +10,38 @@ -