diff --git a/Directory.build.props b/Directory.build.props index fe14bb0e..bd8abd70 100644 --- a/Directory.build.props +++ b/Directory.build.props @@ -2,7 +2,6 @@ $(MSBuildWarningsAsMessages);NETSDK1202 - true diff --git a/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs b/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs index a8fdb65e..3dc08ccb 100644 --- a/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs +++ b/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs @@ -25,3 +25,28 @@ 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. + /// + /// Currently, checkpoint request IDs are represented as strings. This is because some PowerSync SDKs are for runtimes + /// that don't have a fast 64-bit integer type. In a future release, checkpoint request IDs will change to be + /// represented by longs, meaning the parameter's type will also change to `long`. + /// + Task PostCheckpointRequest(string clientId, string requestId); +} 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 3e1e7fd1..17b4b627 100644 --- a/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs +++ b/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs @@ -50,6 +50,13 @@ public class PowerSyncDatabaseOptions() : BasePowerSyncDatabaseOptions() /// If not provided, a default Remote will be created. /// public Func? RemoteFactory { get; set; } + + /// + /// Source of the delays used by the sync client (retry delays, upload throttling). + /// Defaults to ; tests substitute a fake clock so they + /// don't have to wait out real retry delays. + /// + 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) => + { + await WaitForReady(); + return await checkpointConnector.PostCheckpointRequest(clientId, requestId); + } + : null, RetryDelayMs = options.RetryDelayMs, Subscriptions = options.Subscriptions, CrudUploadThrottleMs = options.CrudUploadThrottleMs, + TimeProvider = timeProvider, Logger = Logger }); @@ -451,16 +467,6 @@ public async Task Init() await WaitForReady(); } - private RequiredAdditionalConnectionOptions resolveConnectionOptions(PowerSyncConnectionOptions? options) - { - var defaults = RequiredAdditionalConnectionOptions.DEFAULT_ADDITIONAL_CONNECTION_OPTIONS; - return new RequiredAdditionalConnectionOptions - { - RetryDelayMs = options?.RetryDelayMs ?? defaults.RetryDelayMs, - CrudUploadThrottleMs = options?.CrudUploadThrottleMs ?? defaults.CrudUploadThrottleMs, - }; - } - public async Task Connect(IPowerSyncBackendConnector connector, PowerSyncConnectionOptions? options = null) { await WaitForReady(); @@ -829,6 +835,9 @@ public class SQLWatchOptions /// 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 a02d1341..3ccb512c 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs @@ -6,6 +6,7 @@ namespace PowerSync.Common.Client.Sync.Bucket; using Newtonsoft.Json; +using PowerSync.Common.DB; using PowerSync.Common.DB.Crud; using PowerSync.Common.Utils; @@ -19,8 +20,6 @@ public static class PowerSyncControlCommand public const string NOTIFY_CRUD_UPLOAD_COMPLETED = "completed_upload"; public const string UPDATE_SUBSCRIPTIONS = "update_subscriptions"; - public const string TARGET_CHECKPOINT_REQUEST_ID = "target_checkpoint_request_id"; - /// /// An `established` or `end` event for response streams. /// @@ -136,9 +135,14 @@ public interface IBucketStorageAdapter : ICloseable Task GetCrudBatch(int limit = 100); Task UpdateLocalTarget(Func> callback); - Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null); + // TODO Return int64 from this in future release + /// + /// Reads or updates the local checkpoint request ID counter. + /// + Task ReadOrUpdateCheckpoint(string variant, string? update = null); + /// /// Get a unique client ID. /// @@ -149,3 +153,47 @@ public interface IBucketStorageAdapter : ICloseable /// Task Control(string op, object? payload); } + +/// +/// Provides type-safe wrappers for . +/// +/// Default Interface Implementations would be preferred here, but netstandard2.0 doesn't +/// support them. +/// +public static class BucketStorageAdapterExtensions +{ + /// + /// Increments and returns the local checkpoint counter. + /// + public static Task NextCheckpointRequestId(this IBucketStorageAdapter adapter) + => adapter.ReadOrUpdateCheckpoint("next")!; + + /// + /// Returns the highest checkpoint request ID that has been requested on this device. + /// + public static Task CurrentCheckpointRequestId(this IBucketStorageAdapter adapter) + => adapter.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 static Task SeedCheckpointRequestId(this IBucketStorageAdapter adapter, string serviceResponse) + => adapter.ReadOrUpdateCheckpoint("seed", serviceResponse)!; +} diff --git a/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs b/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs index 0a132c28..2e64404f 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs @@ -10,7 +10,6 @@ namespace PowerSync.Common.Client.Sync.Bucket; using Newtonsoft.Json; -using PowerSync.Common.Client.Sync.Stream; using PowerSync.Common.DB; using PowerSync.Common.DB.Crud; @@ -69,18 +68,26 @@ public async Task 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. /// - /// The previous checkpoint request. - private static Task TargetCheckpointRequestId(ILockContext tx, string? update = null) + public Task ReadOrUpdateCheckpoint(string variant, string? update = null) + => db.WriteTransaction(tx => ReadOrUpdateCheckpoint(tx, variant, update)); + + /// + /// Reads or updates the stored checkpoint request id using the given transaction. + /// + public static Task ReadOrUpdateCheckpoint(ITransaction tx, string variant, string? payload = null) { - // TODO Note that we are only casting in Dart/JS because this returns a 64-bit integer we can't natively represent there. - // Turning MAX_OP_ID into a 64-bit integer here and comparing ints would be better. + // TODO Return 64-bit integer in later release. return tx.Get( "SELECT CAST(powersync_control(?, ?) AS TEXT) AS r", - [PowerSyncControlCommand.TARGET_CHECKPOINT_REQUEST_ID, update]); + [$"{variant}_checkpoint_request_id", payload]); } + // This is called within existing transactions, therefore accept an ITransaction instead of creating a new one + private static Task TargetCheckpointRequestId(ITransaction tx, string? update = null) + => ReadOrUpdateCheckpoint(tx, "target", update); + private record ResultResult(object result); public class ResultDetail @@ -154,6 +161,7 @@ public async Task UpdateLocalTarget(Func> callback) return true; }); } + public Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null) { return db.WriteTransaction(async tx => diff --git a/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs b/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs new file mode 100644 index 00000000..8f99fb46 --- /dev/null +++ b/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs @@ -0,0 +1,23 @@ +namespace PowerSync.Common.Client.Sync; + +/// 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"; +} 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..16f020d2 --- /dev/null +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/CheckpointState.cs @@ -0,0 +1,165 @@ +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 + { + while (!HandleState(wakeDownloadLoop)) + { + 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 HandleState(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..d3c44bc6 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 string? 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..0cb7edf9 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/Remote.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/Remote.cs @@ -102,12 +102,13 @@ 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 client.SendAsync(request, ct); if (response.StatusCode == System.Net.HttpStatusCode.Unauthorized) { @@ -129,8 +130,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 1dbeeccd..a1dfea0c 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(); @@ -167,19 +183,26 @@ public class StreamingSyncImplementation : ICloseable 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. + /// + private string? 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(); + private readonly StreamingSyncLocks locks; public StreamingSyncImplementation(StreamingSyncImplementationOptions options) @@ -200,25 +223,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 +232,6 @@ public StreamingSyncImplementation(StreamingSyncImplementationOptions options) /// public bool IsConnected => SyncStatus.Connected; - /// /// The timestamp of the last successful sync. /// @@ -293,6 +298,10 @@ public async Task Disconnect() await streamingSyncTask; } } + catch (OperationCanceledException) + { + // Expected: disconnecting cancels whatever the sync loops had in flight. + } catch (Exception ex) { // The operation might have failed, all we care about is if it has completed @@ -301,21 +310,71 @@ 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); + } + + /// + /// 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() + ?? throw new InvalidOperationException("The core extension did not return a checkpoint request id."); + var clientId = await Options.Adapter.GetClientId(); + return await RequestCheckpointFromService(signal, new CheckpointRequestPayload { - try - { - await crudUploadTask; - } - catch (Exception ex) - { - logger.LogWarning("CRUD upload task failed during disconnect: {Message}", ex.Message); - } - crudUploadTask = null; + 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); } - UpdateSyncStatus(new SyncStatusOptions { Connected = false, Connecting = false }); + 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); + } + + // TODO convert write checkpoint data type to long in a future release + private async Task GetLegacyWriteCheckpoint() + { + var clientId = await Options.Adapter.GetClientId(); + var path = $"/write-checkpoint2.json?client_id={clientId}"; + var response = await Options.Remote.FetchJson(path); + + logger.LogDebug("Created write checkpoint: {checkpoint}", response.Data.WriteCheckpoint); + return response.Data.WriteCheckpoint; } protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectionOptions? options) @@ -326,6 +385,29 @@ protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectio signal = CancellationTokenSource.Token; } + var token = signal.Value; + var resolvedOptions = options ?? new PowerSyncConnectionOptions(); + + 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,7 +420,7 @@ 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(); @@ -364,7 +446,7 @@ protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectio try { - if (signal.Value.IsCancellationRequested) + if (signal.IsCancellationRequested) { break; } @@ -411,7 +493,7 @@ protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectio { notifyCompletedUploads = null; - if (!signal.Value.IsCancellationRequested) + if (!signal.IsCancellationRequested) { // Closing sync stream network requests before retry. nestedCts.Cancel(); @@ -426,7 +508,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 +523,119 @@ 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 + { + while (!signal.IsCancellationRequested) + { + // Start the initial CRUD upload on connect. Then, keep polling until we're done. + await Task.WhenAll( + InternalUploadAllCrud(signal, options), + 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 = (int)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)) + { + 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); + await RequestCheckpointFromService(signal, new CheckpointRequestPayload + { + ClientId = await Options.Adapter.GetClientId(), + CheckpointRequestId = requestId, + }); + } + 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. + /// + private bool IsCheckpointRequestApplied(string requestId) + { + return lastAppliedCheckpointRequestId is { } applied + && long.TryParse(applied, out var appliedId) + && long.TryParse(requestId, out var required) + && appliedId >= required; + } + protected record StreamingSyncIterationResult { public bool? LegacyRetry { get; init; } @@ -450,8 +647,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 +668,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 +685,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 +816,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 +862,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; @@ -706,6 +914,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 +959,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 +974,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; } @@ -789,6 +1024,16 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) try { await receivingLines; } catch { /* surfaced via streamError */ } } + // 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 { /* surfaced via EnqueuedCommand.Error */ } + } + + // No checkpoint requests can be made until the next iteration seeds its state. + checkpointState.DownloadIterationEnded(); + await Stop(); } @@ -803,25 +1048,8 @@ public void Close() Events.Close(); } - public record ResponseData( - [property: JsonProperty("write_checkpoint")] string 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 +1057,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 +1091,27 @@ 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()); + if (neededUpdate) + { + notifyCompletedUploads?.Invoke(); + } + else if (checkedCrudItem != null) + { + // Only log this if there was something to upload + logger.LogDebug("Upload complete, no write checkpoint needed."); + } break; } } + catch (OperationCanceledException) when (signal.IsCancellationRequested) + { + // Disconnecting. + break; + } catch (Exception ex) { checkedCrudItem = null; @@ -879,7 +1124,7 @@ await locks.ObtainLock(new LockOptions } }); - await DelayRetry(); + await DelayRetry(signal, options.RetryDelayMs ?? DEFAULT_RETRY_DELAY_MS); if (!IsConnected) { @@ -955,22 +1200,122 @@ protected void UpdateSyncStatus(SyncStatusOptions options, UpdateSyncStatusOptio } } - private async Task DelayRetry() + /// + /// Waits out a retry delay. Disconnecting ends the delay rather than throwing: callers check the + /// signal themselves, and a deliberate disconnect isn't a failure worth surfacing. + /// + /// + /// 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 = Options.TimeProvider.Delay(TimeSpan.FromMilliseconds(delay), 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) + { + // Disconnected. + } } - } + // Ends whichever task is still pending. Without this an abandoned checkpoint waiter would + // consume the signal that should have woken the next delay. + nestedCts.Cancel(); + } public void UpdateSubscriptions(SubscribedStream[] subscriptions) { activeStreams = subscriptions; handleActiveStreamsChange?.Invoke(); } + + /// A conflating single-slot channel: only the fact that a signal arrived matters. + private static Channel CreateNotifier() => Channel.CreateBounded(new BoundedChannelOptions(1) { FullMode = BoundedChannelFullMode.DropWrite }); + + internal record LegacyWriteCheckpointResponseData( + [property: JsonProperty("write_checkpoint")] string 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 long MINIMUM_RETRY_DELAY = 10_000; + const long 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 long 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(long 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..4253e155 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs @@ -1,14 +1,15 @@ -namespace PowerSync.Common.Client.Sync.Stream; +using PowerSync.Common.Client.Sync.Bucket; +using PowerSync.Common.DB.Crud; +using PowerSync.Common.Utils.Converters; 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 +96,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; } @@ -131,10 +132,10 @@ public class CheckpointDiff public string 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; } = ""; @@ -182,7 +183,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 +191,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 string CheckpointRequestId { get; set; } +} + +public class CheckpointRequestResponse +{ + [JsonProperty("data")] + public CheckpointRequestResponseData Data { get; set; } +} + +public class CheckpointRequestResponseData +{ + [JsonProperty("checkpoint_request_id")] + public string 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..40c262a0 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; diff --git a/PowerSync/PowerSync.Common/PowerSync.Common.csproj b/PowerSync/PowerSync.Common/PowerSync.Common.csproj index 6e45bb7d..06c02423 100644 --- a/PowerSync/PowerSync.Common/PowerSync.Common.csproj +++ b/PowerSync/PowerSync.Common/PowerSync.Common.csproj @@ -19,6 +19,7 @@ icon.png NU5100 README.md + true $(DefaultItemExcludes);runtimes/**/*.*; @@ -29,6 +30,7 @@ + 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/PowerSync.Maui.csproj b/PowerSync/PowerSync.Maui/PowerSync.Maui.csproj index 74e9a948..6cfb52fb 100644 --- a/PowerSync/PowerSync.Maui/PowerSync.Maui.csproj +++ b/PowerSync/PowerSync.Maui/PowerSync.Maui.csproj @@ -19,6 +19,7 @@ icon.png NU5100 README.md + true true true 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..9fb250f9 --- /dev/null +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs @@ -0,0 +1,318 @@ +using Microsoft.Extensions.Time.Testing; + +using PowerSync.Common.Client; +using PowerSync.Common.Client.Connection; +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); + } + + // 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, string requestId) + { + return Task.FromResult(requestId); + } +} diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs index 659af874..a5ab3597 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs @@ -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 94a00085..e8aa9633 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationControlFlowTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationControlFlowTests.cs @@ -257,6 +257,7 @@ public Task Control(string op, object? payload) public Task GetCrudBatch(int limit = 100) => Task.FromResult(null); public Task UpdateLocalTarget(Func> callback) => Task.FromResult(false); public Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null) => Task.CompletedTask; + public Task ReadOrUpdateCheckpoint(string variant, string? 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..0869ab5a 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs @@ -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/PowerSync.Common.Tests.csproj b/Tests/PowerSync/PowerSync.Common.Tests/PowerSync.Common.Tests.csproj index a153f06b..b14fd3f7 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/PowerSync.Common.Tests.csproj +++ b/Tests/PowerSync/PowerSync.Common.Tests/PowerSync.Common.Tests.csproj @@ -13,6 +13,7 @@ + diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs b/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs index 2169d1d3..2238cc11 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, long.Parse(request.CheckpointRequestId)); + lastWriteCheckpoint = resolved; + } + + await BeforeCheckpointRequestResponse(); + + return new CheckpointRequestResponse + { + Data = new CheckpointRequestResponseData { CheckpointRequestId = resolved.ToString() } + }; + } + 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, string requestId) + => _postCheckpointRequest(clientId, requestId); +} + +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) => 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..6d2eb148 100644 --- a/Tools/Setup/Setup.cs +++ b/Tools/Setup/Setup.cs @@ -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); }