From 718b0646168fbe746d0877b347c08249246e26ce Mon Sep 17 00:00:00 2001 From: LucDeCaf Date: Thu, 27 Aug 2026 08:28:40 +0200 Subject: [PATCH 1/9] block out --- Directory.build.props | 1 - .../Connection/IPowerSyncBackendConnector.cs | 21 ++++ .../Client/ConnectionManager.cs | 9 ++ .../Client/PowerSyncDatabase.cs | 10 -- .../Client/Sync/Stream/CoreInstructions.cs | 9 ++ .../Stream/StreamingSyncImplementation.cs | 91 +++++++++++--- .../Client/Sync/Stream/StreamingSyncTypes.cs | 25 ++++ .../PowerSync.Common/PowerSync.Common.csproj | 1 + .../PowerSync.Maui/PowerSync.Maui.csproj | 1 + .../Client/Sync/CheckpointRequestsTests.cs | 59 +++++++++ .../Client/Sync/SyncStreamsTests.cs | 28 ++--- .../Utils/Sync/MockSyncService.cs | 112 +++++++++++++----- Tools/Setup/Setup.cs | 4 +- 13 files changed, 300 insertions(+), 71 deletions(-) create mode 100644 Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs 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..4488e6d1 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 +{ + /// + /// TODO + /// + Task PostCheckpointRequest(string clientId, long requestId); +} diff --git a/PowerSync/PowerSync.Common/Client/ConnectionManager.cs b/PowerSync/PowerSync.Common/Client/ConnectionManager.cs index 50b6ce7d..6ad136e3 100644 --- a/PowerSync/PowerSync.Common/Client/ConnectionManager.cs +++ b/PowerSync/PowerSync.Common/Client/ConnectionManager.cs @@ -175,6 +175,15 @@ 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."); + } + else + { + Logger.LogWarning($"It didn't work? CheckpointMode is: {PendingConnectionOptions.Options.CheckpointMode}"); + } // 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..1d69f18a 100644 --- a/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs +++ b/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs @@ -451,16 +451,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(); diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs index 2f94ae5d..eecfe91a 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs @@ -61,6 +61,10 @@ public class EstablishSyncStream : Instruction { [JsonProperty("request")] public StreamingSyncRequest Request { get; set; } = null!; + + // TODO Find out how to make it so that the "request" key is not included if null but only for this class + [JsonProperty("request")] + public CheckpointRequestPayload? CheckpointRequest { get; set; } = null!; } public class UpdateSyncStatus : NonInterruptingInstruction @@ -129,6 +133,11 @@ public class CoreSyncStatus [JsonProperty("streams")] public List Streams { get; set; } = []; + + [JsonProperty("internal_last_applied_checkpoint_request_id")] + // TODO Uncomment and implement (copy from temp branch) + // [JsonConverter(typeof(LongToStringConverter))] + public long? LastAppliedCheckpointRequestId { get; set; } } public class SyncPriorityStatus diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs index 1dbeeccd..49f9ad71 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs @@ -16,15 +16,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 +40,6 @@ public class RequiredAdditionalConnectionOptions : AdditionalConnectionOptions public new int CrudUploadThrottleMs { get; set; } public SubscribedStream[] Subscriptions { get; init; } = null!; - } public class StreamingSyncImplementationOptions : AdditionalConnectionOptions @@ -59,7 +55,7 @@ public class StreamingSyncImplementationOptions : AdditionalConnectionOptions public ILogger? Logger { get; init; } } -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 +73,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 +126,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 +148,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(); @@ -466,6 +469,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); @@ -655,7 +659,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 == CheckpointMode.Legacy ? "legacy" : "requests", }; StreamingSyncRequest? establishRequest = null; @@ -971,6 +976,64 @@ public void UpdateSubscriptions(SubscribedStream[] subscriptions) } } +/// +/// 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..30b31560 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs @@ -190,3 +190,28 @@ 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")] + // TODO Uncomment and implement (copy from temp branch) + // [JsonConverter(typeof(LongToStringConverter))] + public long CheckpointRequestId { get; set; } +} + +public class CheckpointRequestResponse +{ + [JsonProperty("data")] + public CheckpointRequestResponseData Data { get; set; } +} + +public class CheckpointRequestResponseData +{ + [JsonProperty("checkpoint_request_id")] + // TODO Uncomment and implement (copy from temp branch) + // [JsonConverter(typeof(LongToStringConverter))] + public long CheckpointRequestId { get; set; } +} diff --git a/PowerSync/PowerSync.Common/PowerSync.Common.csproj b/PowerSync/PowerSync.Common/PowerSync.Common.csproj index 6e45bb7d..9f4e1cc8 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/**/*.*; 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..7ca3158e --- /dev/null +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs @@ -0,0 +1,59 @@ +using PowerSync.Common.Tests.Utils; +using PowerSync.Common.Tests.Utils.Sync; +using PowerSync.Common.Client; +using PowerSync.Common.Client.Connection; +using PowerSync.Common.Client.Sync.Stream; + +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!; + + 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 CheckpointRequestConnector(), new(checkpointMode: new CheckpointMode.Requests())); + + var logs = _syncService.Logs; + Assert.Single(logs); + Assert.Contains("implements ICustomCheckpointRequestConnector, but Connect() was called without checkpoint requests enabled.", logs[0].Message); + } +} + +class CheckpointRequestConnector : TestConnector, ICustomCheckpointRequestConnector +{ + public Task PostCheckpointRequest(string clientId, long requestId) + { + return Task.FromResult(requestId); + } +} diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs index 88a4972f..d95afdf0 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs @@ -40,7 +40,7 @@ public async Task CanDisableDefaultStreams() IncludeDefaultStreams = false }); - TestUtils.DeepEquivalent(new RequestStream { IncludeDefaults = false, Subscriptions = [] }, syncService.Requests[0].Streams); + TestUtils.DeepEquivalent(new RequestStream { IncludeDefaults = false, Subscriptions = [] }, syncService.StreamingSyncRequests[0].Streams); } [Fact] @@ -49,8 +49,8 @@ public async Task BasicSubscribeTest() var a = await db.SyncStream("a").Subscribe(); await db.Connect(new TestConnector(), new PowerSyncConnectionOptions()); - Assert.Equal(1, syncService.Requests[0]?.Streams?.Subscriptions.Count); - Assert.Equal("a", syncService.Requests[0]?.Streams?.Subscriptions[0].Stream); + Assert.Equal(1, syncService.StreamingSyncRequests[0]?.Streams?.Subscriptions.Count); + Assert.Equal("a", syncService.StreamingSyncRequests[0]?.Streams?.Subscriptions[0].Stream); a.Unsubscribe(); } @@ -64,8 +64,8 @@ public async Task SubscribesWithStreams() await db.Connect(new TestConnector()); - Assert.True(syncService.Requests[0]?.Streams?.IncludeDefaults); - Assert.Equal(2, syncService.Requests[0]?.Streams?.Subscriptions.Count); + Assert.True(syncService.StreamingSyncRequests[0]?.Streams?.IncludeDefaults); + Assert.Equal(2, syncService.StreamingSyncRequests[0]?.Streams?.Subscriptions.Count); TestUtils.DeepEquivalent( new RequestStreamSubscription { @@ -73,7 +73,7 @@ public async Task SubscribesWithStreams() Parameters = new Dictionary { { "foo", "a" } }, OverridePriority = null }, - syncService.Requests[0]?.Streams?.Subscriptions[0] + syncService.StreamingSyncRequests[0]?.Streams?.Subscriptions[0] ); TestUtils.DeepEquivalent( new RequestStreamSubscription @@ -82,7 +82,7 @@ public async Task SubscribesWithStreams() Parameters = new Dictionary { { "foo", "b" } }, OverridePriority = 1 }, - syncService.Requests[0]?.Streams?.Subscriptions[1] + syncService.StreamingSyncRequests[0]?.Streams?.Subscriptions[1] ); var statusTask = MockSyncService.NextStatus(db); @@ -159,13 +159,13 @@ public async Task ChangesSubscriptionsDynamically() var subscription = await db.SyncStream("a").Subscribe(); // Wait for subscription request to register - await TestUtils.WaitForAsync(() => syncService.Requests.Count > 1); - Assert.Single(syncService.Requests[1]?.Streams?.Subscriptions!); + await TestUtils.WaitForAsync(() => syncService.StreamingSyncRequests.Count > 1); + Assert.Single(syncService.StreamingSyncRequests[1]?.Streams?.Subscriptions!); // Given that the subscription has a TTL, dropping the handle should not re-subscribe. subscription.Unsubscribe(); - await TestUtils.WaitForAsync(() => syncService.Requests.Count == 2); - Assert.Equal(2, syncService.Requests.Count); + await TestUtils.WaitForAsync(() => syncService.StreamingSyncRequests.Count == 2); + Assert.Equal(2, syncService.StreamingSyncRequests.Count); } [Fact] @@ -192,8 +192,8 @@ public async Task UnsubscribeMultipleTimesHasNoEffectTest() await db.Execute("UPDATE ps_stream_subscriptions SET expires_at = unixepoch() - 1000"); await db.Connect(new TestConnector()); - Assert.True(syncService.Requests[0]?.Streams?.IncludeDefaults); - Assert.Single(syncService.Requests[0]?.Streams?.Subscriptions!); + Assert.True(syncService.StreamingSyncRequests[0]?.Streams?.IncludeDefaults); + Assert.Single(syncService.StreamingSyncRequests[0]?.Streams?.Subscriptions!); aAgain.Unsubscribe(); } @@ -205,6 +205,6 @@ public async Task UnsubscribeAllTest() await db.SyncStream("a").UnsubscribeAll(); await db.Connect(new TestConnector(), new PowerSyncConnectionOptions()); - TestUtils.DeepEquivalent(new RequestStream { IncludeDefaults = true, Subscriptions = [] }, syncService.Requests[0].Streams); + TestUtils.DeepEquivalent(new RequestStream { IncludeDefaults = true, Subscriptions = [] }, syncService.StreamingSyncRequests[0].Streams); } } diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs b/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs index 2169d1d3..f7df9343 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,17 @@ 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; + + public long LastWriteCheckpoint { get; set; } = 0; + + private readonly List _checkpointRequests = []; + public IReadOnlyList CheckpointRequests => _checkpointRequests; + public void PushLine(StreamingSyncLine line) { Emit(JsonConvert.SerializeObject(line)); @@ -44,16 +51,17 @@ public PowerSyncDatabase CreateDatabase(string? dbFilename = null) Database = new SQLOpenOptions { DbFilename = dbFilename }, Schema = TestSchemaTodoList.AppSchema, RemoteFactory = _ => mockRemote, - Logger = createLogger() + 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"); } @@ -154,43 +162,58 @@ public MockRemote( this.connectedListeners = connectedListeners; } + // TODO This should be able to parse and handle /sync/stream AND /sync/checkpoint_request (or whatever the URL is) 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()); + } + else if (options.Path.Contains("/sync/checkpoint-request")) + { + + } + else if (options.Path.Contains("/write-checkpoint2.json")) + { + } } public override Task Get(string path, Dictionary? headers = null) { - var response = new StreamingSyncImplementation.ApiResponse( - new StreamingSyncImplementation.ResponseData("1") - ); + // Write checkpoint + if (path.Contains("checkpoint2.json")) + { + return Task.FromResult(new StreamingSyncImplementation.ApiResponse( + new StreamingSyncImplementation.ResponseData("1") + )); + } - return Task.FromResult((T)(object)response); + throw new InvalidOperationException("Not implemented"); } } @@ -213,3 +236,32 @@ public async Task UploadData(IPowerSyncDatabase database) } } } + +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); } From 3e84a59a1b84e85b8ce3d25ff608fbe4e941ca2d Mon Sep 17 00:00:00 2001 From: LucDeCaf Date: Thu, 27 Aug 2026 11:57:45 +0200 Subject: [PATCH 2/9] more partial work --- ; | 234 ++++++++++++++++++ .../Client/ConnectionManager.cs | 4 - .../Client/PowerSyncDatabase.cs | 3 + .../Sync/Bucket/BucketStorageAdapter.cs | 3 +- .../Client/Sync/Bucket/SqliteBucketStorage.cs | 10 + .../Client/Sync/Stream/CoreInstructions.cs | 9 +- .../Client/Sync/Stream/Remote.cs | 19 +- .../Stream/StreamingSyncImplementation.cs | 70 ++++-- .../Client/Sync/Stream/StreamingSyncTypes.cs | 20 +- .../PowerSync.Common/DB/Crud/SyncProgress.cs | 6 +- .../PowerSync.Common/DB/Crud/SyncStatus.cs | 2 - .../Utils/Converters/LongToStringConverter.cs | 43 ++++ .../Client/Sync/CheckpointRequestsTests.cs | 51 +++- .../Client/Sync/StreamingSyncRetryTests.cs | 2 +- .../Client/Sync/SyncStreamsTests.cs | 28 +-- .../Utils/Sync/MockSyncService.cs | 15 +- 16 files changed, 441 insertions(+), 78 deletions(-) create mode 100644 ; create mode 100644 PowerSync/PowerSync.Common/Utils/Converters/LongToStringConverter.cs diff --git a/; b/; new file mode 100644 index 00000000..f3dde77f --- /dev/null +++ b/; @@ -0,0 +1,234 @@ +namespace PowerSync.Common.Client.Sync.Bucket; + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; + +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; + +using Newtonsoft.Json; + +using PowerSync.Common.Client.Sync.Stream; +using PowerSync.Common.DB; +using PowerSync.Common.DB.Crud; + +public class SqliteBucketStorage : IBucketStorageAdapter +{ + public static readonly string MAX_OP_ID = "9223372036854775807"; + + public BucketStorageEvents Events { get; } = new(); + + private readonly IDBAdapter db; + + private string? clientId; + + private readonly ILogger logger; + + private readonly CancellationTokenSource updateCts; + private readonly Task updateTask; + + public SqliteBucketStorage(IDBAdapter db, ILogger? logger = null) + { + this.db = db; + this.logger = logger ?? NullLogger.Instance; + + updateCts = new CancellationTokenSource(); + + updateTask = Task.Run(() => + { + foreach (var update in db.Events.OnTablesUpdated.Listen(updateCts.Token)) + { + var tables = DBAdapterUtils.ExtractTableUpdates(update.TablesUpdated); + if (tables.Contains(PSInternalTable.CRUD)) + { + Events.Emit(new BucketStorageEvents.CrudUpdateEvent()); + } + } + }); + } + + public void Close() + { + updateCts.Cancel(); + try { updateTask.Wait(2000); } catch (Exception) { } + Events.Close(); + } + + private record ClientIdResult(string? client_id); + public async Task GetClientId() + { + if (clientId == null) + { + var row = await db.Get("SELECT powersync_client_id() as client_id"); + clientId = row.client_id ?? ""; + } + + return clientId; + } + + /// + /// Reads the stored target checkpoint request id, or updates it when the update parameter is set. + /// + /// The previous checkpoint request. + private static Task TargetCheckpointRequestId(ILockContext tx, string? update = 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. + return tx.Get( + "SELECT CAST(powersync_control(?, ?) AS TEXT) AS r", + [PowerSyncControlCommand.TARGET_CHECKPOINT_REQUEST_ID, update]); + } + + private record ResultResult(object result); + + public class ResultDetail + { + [JsonProperty("valid")] + public bool Valid { get; set; } + + [JsonProperty("failed_buckets")] + public List? FailedBuckets { get; set; } + } + + private record SequenceResult(long seq); + + public async Task UpdateLocalTarget(Func> callback) + { + var seqBeforeResult = await db.ReadTransaction(async tx => + { + var currentTarget = await TargetCheckpointRequestId(tx); + if (currentTarget != MAX_OP_ID) + { + // Nothing to update + return (long?)null; + } + + var rs = await tx.GetAll( + "SELECT seq FROM main.sqlite_sequence WHERE name = 'ps_crud'" + ); + + return rs.Length == 0 ? null : rs[0].seq; + }); + + if (seqBeforeResult is not { } seqBefore) + { + // Nothing to update + return false; + } + + string opId = await callback(); + + logger.LogDebug("[updateLocalTarget] Updating target to checkpoint {message}", opId); + + return await db.WriteTransaction(async tx => + { + var anyData = await tx.Execute("SELECT 1 FROM ps_crud LIMIT 1"); + if (anyData.RowsAffected > 0) + { + logger.LogDebug("[updateLocalTarget] ps crud is not empty"); + return false; + } + + var rsAfter = await tx.GetAll( + "SELECT seq FROM main.sqlite_sequence WHERE name = 'ps_crud'" + ); + + if (rsAfter.Length == 0) + { + throw new Exception("SQLite Sequence should not be empty"); + } + + long seqAfter = rsAfter[0].seq; + logger.LogDebug("[updateLocalTarget] seqAfter: {seq}", seqAfter); + + if (seqAfter != seqBefore) + { + logger.LogDebug("[updateLocalTarget] seqAfter ({seqAfter}) != seqBefore ({seqBefore})", seqAfter, + seqBefore); + return false; + } + + await TargetCheckpointRequestId(tx, opId); + return true; + }); + } + public Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null) + { + return db.WriteTransaction(async tx => + { + await tx.Execute($"DELETE FROM {PSInternalTable.CRUD} WHERE id <= ?", [lastClientId]); + + var crudRemaining = await tx.GetOptional( + $"SELECT 1 as ignore FROM {PSInternalTable.CRUD} LIMIT 1") != null; + + await TargetCheckpointRequestId( + tx, + !string.IsNullOrEmpty(writeCheckpoint) && !crudRemaining ? writeCheckpoint : MAX_OP_ID); + }); + } + + /// + /// Get a batch of objects to send to the server. + /// When the objects are successfully sent to the server, call .Complete(). + /// + public async Task GetCrudBatch(int limit = 100) + { + if (!await HasCrud()) + { + return null; + } + + var crudResult = await db.GetAll("SELECT * FROM ps_crud ORDER BY id ASC LIMIT ?", [limit]); + + var all = crudResult.Select(CrudEntry.FromRow).ToArray(); + + if (all.Length == 0) + { + return null; + } + + var last = all[all.Length - 1]; + + return new CrudBatch( + Crud: all, + HaveMore: true, + CompleteCallback: writeCheckpoint => HandleCrudCheckpoint(last.ClientId, writeCheckpoint) + ); + } + + public async Task PostCheckpointRequestId(string variant, long? payload = null) + { + // TODO validate Control this happens in transaction + string rawResponse = await Control($"{variant}_checkpoint_request_id", payload); + if (long.TryParse(rawResponse, out var response)) + { + return response; + } + throw new Exception($"Expected {variant}_checkpoint_request_id to return value of type long."); + } + + public async Task NextCrudItem() + { + var next = await db.GetOptional("SELECT * FROM ps_crud ORDER BY id ASC LIMIT 1"); + + return next != null ? CrudEntry.FromRow(next) : null; + } + + public async Task HasCrud() + { + return await db.GetOptional("SELECT 1 as ignore FROM ps_crud LIMIT 1") != null; + } + + private record ControlResult(string? r); + + public async Task Control(string op, object? payload = null) + { + return await db.WriteTransaction(async tx => + { + var result = await tx.Get("SELECT powersync_control(?, ?) AS r", [op, payload]); + return result.r!; + }); + } +} diff --git a/PowerSync/PowerSync.Common/Client/ConnectionManager.cs b/PowerSync/PowerSync.Common/Client/ConnectionManager.cs index 6ad136e3..4b86f952 100644 --- a/PowerSync/PowerSync.Common/Client/ConnectionManager.cs +++ b/PowerSync/PowerSync.Common/Client/ConnectionManager.cs @@ -180,10 +180,6 @@ public async Task Connect(IPowerSyncBackendConnector connector, PowerSyncConnect { Logger.LogWarning("The backend connector implements ICustomCheckpointRequestConnector, but Connect() was called without checkpoint requests enabled."); } - else - { - Logger.LogWarning($"It didn't work? CheckpointMode is: {PendingConnectionOptions.Options.CheckpointMode}"); - } // 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 1d69f18a..36faa46a 100644 --- a/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs +++ b/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs @@ -819,6 +819,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..151dace0 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs @@ -136,8 +136,9 @@ public interface IBucketStorageAdapter : ICloseable Task GetCrudBatch(int limit = 100); Task UpdateLocalTarget(Func> callback); - Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null); + // TODO type safety + Task ReadCheckpointRequestId(string variant, string? payload = null); /// /// Get a unique client ID. diff --git a/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs b/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs index 0a132c28..da3e52dd 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs @@ -198,6 +198,16 @@ await TargetCheckpointRequestId( ); } + public async Task PostCheckpointRequestId(string variant, long? payload = null) + { + string rawResponse = await Control($"{variant}_checkpoint_request_id", payload); + if (long.TryParse(rawResponse, out var response)) + { + return response; + } + throw new Exception($"Expected {variant}_checkpoint_request_id to return value of type long."); + } + public async Task NextCrudItem() { var next = await db.GetOptional("SELECT * FROM ps_crud ORDER BY id ASC LIMIT 1"); diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs index eecfe91a..128f62ff 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) @@ -62,8 +63,7 @@ public class EstablishSyncStream : Instruction [JsonProperty("request")] public StreamingSyncRequest Request { get; set; } = null!; - // TODO Find out how to make it so that the "request" key is not included if null but only for this class - [JsonProperty("request")] + [JsonProperty("checkpoint_request", NullValueHandling = NullValueHandling.Ignore)] public CheckpointRequestPayload? CheckpointRequest { get; set; } = null!; } @@ -135,8 +135,7 @@ public class CoreSyncStatus public List Streams { get; set; } = []; [JsonProperty("internal_last_applied_checkpoint_request_id")] - // TODO Uncomment and implement (copy from temp branch) - // [JsonConverter(typeof(LongToStringConverter))] + [JsonConverter(typeof(LongToStringConverter))] public long? LastAppliedCheckpointRequestId { get; set; } } diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/Remote.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/Remote.cs index c253d34d..6b955a3d 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) { @@ -116,11 +117,11 @@ public virtual async Task Get(string path, Dictionary? hea if (!response.IsSuccessStatusCode) { - var errorMessage = await response.Content.ReadAsStringAsync(); + var errorMessage = await response.Content.ReadAsStringAsync(ct); throw new HttpRequestException($"Received {response.StatusCode} - {response.ReasonPhrase} when getting from {path}: {errorMessage}"); } - var responseData = await response.Content.ReadAsStringAsync(); + var responseData = await response.Content.ReadAsStringAsync(ct); return JsonConvert.DeserializeObject(responseData)!; } @@ -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) { @@ -144,11 +145,11 @@ public virtual async Task PostStreamRaw(SyncStreamOptions options) if (!response.IsSuccessStatusCode) { - var errorText = await response.Content.ReadAsStringAsync(); + var errorText = await response.Content.ReadAsStringAsync(options.CancellationToken); throw new HttpRequestException($"HTTP {response.StatusCode}: {errorText}"); } - return await response.Content.ReadAsStreamAsync(); + return await response.Content.ReadAsStreamAsync(options.CancellationToken); } private async Task BuildRequest(HttpMethod method, string path, object? data = null, Dictionary? additionalHeaders = null) diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs index 49f9ad71..2396be50 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs @@ -222,6 +222,7 @@ public StreamingSyncImplementation(StreamingSyncImplementationOptions options) isUploadingCrud = false; }); }; + } /// @@ -321,6 +322,48 @@ public async Task Disconnect() UpdateSyncStatus(new SyncStatusOptions { Connected = false, Connecting = false }); } + private async Task RequestNextCheckpointFromService(CancellationToken signal) + { + // TODO CheckpointState manager + await Checkpoints.WaitForCheckpointRequestsReady(signal); + + // TODO implement on adapter + // TODO type safety + var nextCheckpointRequestId = await Options.Adapter.ReadCheckpointRequestId("next"); + var clientId = await Options.Adapter.GetClientId(); + return await RequestCheckpointFromService(signal, new() + { + ClientId = clientId, + CheckpointRequestId = nextCheckpointRequestId, + }); + } + + private async Task RequestCheckpointFromService(CancellationToken signal, CheckpointRequestPayload request) + { + // First, check if we can use a custom checkpoint request implementation. + // TODO add and default implement PostCheckpointRequest + var customResponse = await Options.PostCheckpointRequest(request.ClientId, request.CheckpointRequestId); + if (customResponse != null) return customResponse; + + var status = await Options.Remote.FetchJson( + path: "/sync/checkpoint-request", + method: HttpMethod.Post, + data: request, + ct: signal + ); + return status.Data.CheckpointRequestId; + } + + // TODO convert write checkpoint data type to long + 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); + + return response.Data.WriteCheckpoint; + } + protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectionOptions? options) { if (signal == null) @@ -455,7 +498,6 @@ protected record EnqueuedCommand public object? Payload { get; init; } } - protected async Task StreamingSyncIteration(CancellationToken signal, PowerSyncConnectionOptions? options) { return await locks.ObtainLock(new LockOptions @@ -808,22 +850,6 @@ 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() - { - 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() { @@ -868,7 +894,7 @@ await locks.ObtainLock(new LockOptions else { // Uploading is completed - await Options.Adapter.UpdateLocalTarget(GetWriteCheckpoint); + await Options.Adapter.UpdateLocalTarget(GetLegacyWriteCheckpoint); break; } } @@ -968,12 +994,18 @@ private async Task DelayRetry() } } - public void UpdateSubscriptions(SubscribedStream[] subscriptions) { activeStreams = subscriptions; handleActiveStreamsChange?.Invoke(); } + + public record LegacyWriteCheckpointResponseData( + [property: JsonProperty("write_checkpoint")] string WriteCheckpoint + ); + public record LegacyWriteCheckpointApiResponse( + [property: JsonProperty("data")] LegacyWriteCheckpointResponseData Data + ); } /// diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs index 30b31560..18a1b372 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 @@ -197,8 +198,7 @@ public class CheckpointRequestPayload public string ClientId { get; set; } [JsonProperty("checkpoint_request_id")] - // TODO Uncomment and implement (copy from temp branch) - // [JsonConverter(typeof(LongToStringConverter))] + [JsonConverter(typeof(LongToStringConverter))] 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..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/Utils/Converters/LongToStringConverter.cs b/PowerSync/PowerSync.Common/Utils/Converters/LongToStringConverter.cs new file mode 100644 index 00000000..92eabf86 --- /dev/null +++ b/PowerSync/PowerSync.Common/Utils/Converters/LongToStringConverter.cs @@ -0,0 +1,43 @@ +using Newtonsoft.Json; + +namespace PowerSync.Common.Utils.Converters; + +/// +/// Converts a long to a string representation when converting JSON values. Used +/// for converting checkpoint request IDs from a long to a string before being +/// passed to the core extension. +/// +internal class LongToStringConverter : 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/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs index 7ca3158e..3d84440f 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs @@ -3,6 +3,7 @@ using PowerSync.Common.Client; using PowerSync.Common.Client.Connection; using PowerSync.Common.Client.Sync.Stream; +using PowerSync.Common.Client.Sync.Bucket; namespace PowerSync.Common.Tests.Client.Sync; @@ -29,7 +30,7 @@ public async Task DisposeAsync() DatabaseUtils.CleanDb(_db.Database.Name); } - [Fact(Timeout = 15000)] + [Fact] public async Task CheckpointRequests_WarnsCustomConnectorWithoutRequestsEnabled() { await _db.Connect(new CheckpointRequestConnector()); @@ -39,15 +40,55 @@ public async Task CheckpointRequests_WarnsCustomConnectorWithoutRequestsEnabled( Assert.Contains("implements ICustomCheckpointRequestConnector, but Connect() was called without checkpoint requests enabled.", logs[0].Message); } - [Fact(Timeout = 15000)] + [Fact] public async Task CheckpointRequests_RequestsCheckpointsForUpdates() { await _db.Connect(new CheckpointRequestConnector(), new(checkpointMode: new CheckpointMode.Requests())); - var logs = _syncService.Logs; - Assert.Single(logs); - Assert.Contains("implements ICustomCheckpointRequestConnector, but Connect() was called without checkpoint requests enabled.", logs[0].Message); + 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. + await TestUtils.WaitForAsync(() => _syncService.CheckpointRequests.Count == 2); + + _syncService.PushLine(new StreamingSyncCheckpoint + { + Checkpoint = new() + { + LastOpId = "1", + Buckets = [new() { Bucket = "a", Count = 1, Checksum = 0, Priority = 3 }], + WriteCheckpoint = _syncService.LastWriteCheckpoint.ToString(), + } + }); + _syncService.PushLine(new StreamingSyncDataJSON + { + Data = new() + { + 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); } + private record NameResult(string name); } class CheckpointRequestConnector : TestConnector, ICustomCheckpointRequestConnector diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs index 659af874..5b390cc3 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs @@ -96,7 +96,7 @@ 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") diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs index d95afdf0..88a4972f 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs @@ -40,7 +40,7 @@ public async Task CanDisableDefaultStreams() IncludeDefaultStreams = false }); - TestUtils.DeepEquivalent(new RequestStream { IncludeDefaults = false, Subscriptions = [] }, syncService.StreamingSyncRequests[0].Streams); + TestUtils.DeepEquivalent(new RequestStream { IncludeDefaults = false, Subscriptions = [] }, syncService.Requests[0].Streams); } [Fact] @@ -49,8 +49,8 @@ public async Task BasicSubscribeTest() var a = await db.SyncStream("a").Subscribe(); await db.Connect(new TestConnector(), new PowerSyncConnectionOptions()); - Assert.Equal(1, syncService.StreamingSyncRequests[0]?.Streams?.Subscriptions.Count); - Assert.Equal("a", syncService.StreamingSyncRequests[0]?.Streams?.Subscriptions[0].Stream); + Assert.Equal(1, syncService.Requests[0]?.Streams?.Subscriptions.Count); + Assert.Equal("a", syncService.Requests[0]?.Streams?.Subscriptions[0].Stream); a.Unsubscribe(); } @@ -64,8 +64,8 @@ public async Task SubscribesWithStreams() await db.Connect(new TestConnector()); - Assert.True(syncService.StreamingSyncRequests[0]?.Streams?.IncludeDefaults); - Assert.Equal(2, syncService.StreamingSyncRequests[0]?.Streams?.Subscriptions.Count); + Assert.True(syncService.Requests[0]?.Streams?.IncludeDefaults); + Assert.Equal(2, syncService.Requests[0]?.Streams?.Subscriptions.Count); TestUtils.DeepEquivalent( new RequestStreamSubscription { @@ -73,7 +73,7 @@ public async Task SubscribesWithStreams() Parameters = new Dictionary { { "foo", "a" } }, OverridePriority = null }, - syncService.StreamingSyncRequests[0]?.Streams?.Subscriptions[0] + syncService.Requests[0]?.Streams?.Subscriptions[0] ); TestUtils.DeepEquivalent( new RequestStreamSubscription @@ -82,7 +82,7 @@ public async Task SubscribesWithStreams() Parameters = new Dictionary { { "foo", "b" } }, OverridePriority = 1 }, - syncService.StreamingSyncRequests[0]?.Streams?.Subscriptions[1] + syncService.Requests[0]?.Streams?.Subscriptions[1] ); var statusTask = MockSyncService.NextStatus(db); @@ -159,13 +159,13 @@ public async Task ChangesSubscriptionsDynamically() var subscription = await db.SyncStream("a").Subscribe(); // Wait for subscription request to register - await TestUtils.WaitForAsync(() => syncService.StreamingSyncRequests.Count > 1); - Assert.Single(syncService.StreamingSyncRequests[1]?.Streams?.Subscriptions!); + await TestUtils.WaitForAsync(() => syncService.Requests.Count > 1); + Assert.Single(syncService.Requests[1]?.Streams?.Subscriptions!); // Given that the subscription has a TTL, dropping the handle should not re-subscribe. subscription.Unsubscribe(); - await TestUtils.WaitForAsync(() => syncService.StreamingSyncRequests.Count == 2); - Assert.Equal(2, syncService.StreamingSyncRequests.Count); + await TestUtils.WaitForAsync(() => syncService.Requests.Count == 2); + Assert.Equal(2, syncService.Requests.Count); } [Fact] @@ -192,8 +192,8 @@ public async Task UnsubscribeMultipleTimesHasNoEffectTest() await db.Execute("UPDATE ps_stream_subscriptions SET expires_at = unixepoch() - 1000"); await db.Connect(new TestConnector()); - Assert.True(syncService.StreamingSyncRequests[0]?.Streams?.IncludeDefaults); - Assert.Single(syncService.StreamingSyncRequests[0]?.Streams?.Subscriptions!); + Assert.True(syncService.Requests[0]?.Streams?.IncludeDefaults); + Assert.Single(syncService.Requests[0]?.Streams?.Subscriptions!); aAgain.Unsubscribe(); } @@ -205,6 +205,6 @@ public async Task UnsubscribeAllTest() await db.SyncStream("a").UnsubscribeAll(); await db.Connect(new TestConnector(), new PowerSyncConnectionOptions()); - TestUtils.DeepEquivalent(new RequestStream { IncludeDefaults = true, Subscriptions = [] }, syncService.StreamingSyncRequests[0].Streams); + TestUtils.DeepEquivalent(new RequestStream { IncludeDefaults = true, Subscriptions = [] }, syncService.Requests[0].Streams); } } diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs b/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs index f7df9343..1d84c77b 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs @@ -196,21 +196,26 @@ public override Task PostStreamRaw(SyncStreamOptions options) } else if (options.Path.Contains("/sync/checkpoint-request")) { - + // TODO + throw new Exception("Not implemented"); } else if (options.Path.Contains("/write-checkpoint2.json")) { + // TODO + throw new Exception("Not implemented"); } + + throw new Exception("Not implemented"); } - 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) { - // Write checkpoint if (path.Contains("checkpoint2.json")) { - return Task.FromResult(new StreamingSyncImplementation.ApiResponse( + var response = (T)(object)new StreamingSyncImplementation.ApiResponse( new StreamingSyncImplementation.ResponseData("1") - )); + ); + return Task.FromResult(response); } throw new InvalidOperationException("Not implemented"); From bbf39f2140f8a1b5ad22afe67fbf5a597ff4e908 Mon Sep 17 00:00:00 2001 From: LucDeCaf Date: Thu, 27 Aug 2026 13:07:00 +0200 Subject: [PATCH 3/9] delete \; --- ; | 234 -------------------------------------------------------------- 1 file changed, 234 deletions(-) delete mode 100644 ; diff --git a/; b/; deleted file mode 100644 index f3dde77f..00000000 --- a/; +++ /dev/null @@ -1,234 +0,0 @@ -namespace PowerSync.Common.Client.Sync.Bucket; - -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading.Tasks; - -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Logging.Abstractions; - -using Newtonsoft.Json; - -using PowerSync.Common.Client.Sync.Stream; -using PowerSync.Common.DB; -using PowerSync.Common.DB.Crud; - -public class SqliteBucketStorage : IBucketStorageAdapter -{ - public static readonly string MAX_OP_ID = "9223372036854775807"; - - public BucketStorageEvents Events { get; } = new(); - - private readonly IDBAdapter db; - - private string? clientId; - - private readonly ILogger logger; - - private readonly CancellationTokenSource updateCts; - private readonly Task updateTask; - - public SqliteBucketStorage(IDBAdapter db, ILogger? logger = null) - { - this.db = db; - this.logger = logger ?? NullLogger.Instance; - - updateCts = new CancellationTokenSource(); - - updateTask = Task.Run(() => - { - foreach (var update in db.Events.OnTablesUpdated.Listen(updateCts.Token)) - { - var tables = DBAdapterUtils.ExtractTableUpdates(update.TablesUpdated); - if (tables.Contains(PSInternalTable.CRUD)) - { - Events.Emit(new BucketStorageEvents.CrudUpdateEvent()); - } - } - }); - } - - public void Close() - { - updateCts.Cancel(); - try { updateTask.Wait(2000); } catch (Exception) { } - Events.Close(); - } - - private record ClientIdResult(string? client_id); - public async Task GetClientId() - { - if (clientId == null) - { - var row = await db.Get("SELECT powersync_client_id() as client_id"); - clientId = row.client_id ?? ""; - } - - return clientId; - } - - /// - /// Reads the stored target checkpoint request id, or updates it when the update parameter is set. - /// - /// The previous checkpoint request. - private static Task TargetCheckpointRequestId(ILockContext tx, string? update = 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. - return tx.Get( - "SELECT CAST(powersync_control(?, ?) AS TEXT) AS r", - [PowerSyncControlCommand.TARGET_CHECKPOINT_REQUEST_ID, update]); - } - - private record ResultResult(object result); - - public class ResultDetail - { - [JsonProperty("valid")] - public bool Valid { get; set; } - - [JsonProperty("failed_buckets")] - public List? FailedBuckets { get; set; } - } - - private record SequenceResult(long seq); - - public async Task UpdateLocalTarget(Func> callback) - { - var seqBeforeResult = await db.ReadTransaction(async tx => - { - var currentTarget = await TargetCheckpointRequestId(tx); - if (currentTarget != MAX_OP_ID) - { - // Nothing to update - return (long?)null; - } - - var rs = await tx.GetAll( - "SELECT seq FROM main.sqlite_sequence WHERE name = 'ps_crud'" - ); - - return rs.Length == 0 ? null : rs[0].seq; - }); - - if (seqBeforeResult is not { } seqBefore) - { - // Nothing to update - return false; - } - - string opId = await callback(); - - logger.LogDebug("[updateLocalTarget] Updating target to checkpoint {message}", opId); - - return await db.WriteTransaction(async tx => - { - var anyData = await tx.Execute("SELECT 1 FROM ps_crud LIMIT 1"); - if (anyData.RowsAffected > 0) - { - logger.LogDebug("[updateLocalTarget] ps crud is not empty"); - return false; - } - - var rsAfter = await tx.GetAll( - "SELECT seq FROM main.sqlite_sequence WHERE name = 'ps_crud'" - ); - - if (rsAfter.Length == 0) - { - throw new Exception("SQLite Sequence should not be empty"); - } - - long seqAfter = rsAfter[0].seq; - logger.LogDebug("[updateLocalTarget] seqAfter: {seq}", seqAfter); - - if (seqAfter != seqBefore) - { - logger.LogDebug("[updateLocalTarget] seqAfter ({seqAfter}) != seqBefore ({seqBefore})", seqAfter, - seqBefore); - return false; - } - - await TargetCheckpointRequestId(tx, opId); - return true; - }); - } - public Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null) - { - return db.WriteTransaction(async tx => - { - await tx.Execute($"DELETE FROM {PSInternalTable.CRUD} WHERE id <= ?", [lastClientId]); - - var crudRemaining = await tx.GetOptional( - $"SELECT 1 as ignore FROM {PSInternalTable.CRUD} LIMIT 1") != null; - - await TargetCheckpointRequestId( - tx, - !string.IsNullOrEmpty(writeCheckpoint) && !crudRemaining ? writeCheckpoint : MAX_OP_ID); - }); - } - - /// - /// Get a batch of objects to send to the server. - /// When the objects are successfully sent to the server, call .Complete(). - /// - public async Task GetCrudBatch(int limit = 100) - { - if (!await HasCrud()) - { - return null; - } - - var crudResult = await db.GetAll("SELECT * FROM ps_crud ORDER BY id ASC LIMIT ?", [limit]); - - var all = crudResult.Select(CrudEntry.FromRow).ToArray(); - - if (all.Length == 0) - { - return null; - } - - var last = all[all.Length - 1]; - - return new CrudBatch( - Crud: all, - HaveMore: true, - CompleteCallback: writeCheckpoint => HandleCrudCheckpoint(last.ClientId, writeCheckpoint) - ); - } - - public async Task PostCheckpointRequestId(string variant, long? payload = null) - { - // TODO validate Control this happens in transaction - string rawResponse = await Control($"{variant}_checkpoint_request_id", payload); - if (long.TryParse(rawResponse, out var response)) - { - return response; - } - throw new Exception($"Expected {variant}_checkpoint_request_id to return value of type long."); - } - - public async Task NextCrudItem() - { - var next = await db.GetOptional("SELECT * FROM ps_crud ORDER BY id ASC LIMIT 1"); - - return next != null ? CrudEntry.FromRow(next) : null; - } - - public async Task HasCrud() - { - return await db.GetOptional("SELECT 1 as ignore FROM ps_crud LIMIT 1") != null; - } - - private record ControlResult(string? r); - - public async Task Control(string op, object? payload = null) - { - return await db.WriteTransaction(async tx => - { - var result = await tx.Get("SELECT powersync_control(?, ?) AS r", [op, payload]); - return result.r!; - }); - } -} From 51263494256588179fc3053bbe5c735700978221 Mon Sep 17 00:00:00 2001 From: LucDeCaf Date: Thu, 27 Aug 2026 17:34:36 +0200 Subject: [PATCH 4/9] Checkpoint state, bucket storage adapter, build no longer errors --- .../Connection/IPowerSyncBackendConnector.cs | 4 +- .../Client/PowerSyncDatabase.cs | 5 +- .../Sync/Bucket/BucketStorageAdapter.cs | 7 +- .../Client/Sync/Bucket/SqliteBucketStorage.cs | 42 +++--- .../Client/Sync/CheckpointRequest.cs | 22 ++++ .../Client/Sync/Stream/CheckpointState.cs | 120 ++++++++++++++++++ .../Client/Sync/Stream/CoreInstructions.cs | 3 +- .../Client/Sync/Stream/Remote.cs | 8 +- .../Stream/StreamingSyncImplementation.cs | 25 ++-- .../Client/Sync/Stream/StreamingSyncTypes.cs | 7 +- .../Utils/BroadcastChannel.cs | 47 +++++++ ...ingConverter.cs => StringLongConverter.cs} | 8 +- .../Client/Sync/StreamingSyncRetryTests.cs | 4 +- .../Utils/Sync/MockSyncService.cs | 14 +- 14 files changed, 264 insertions(+), 52 deletions(-) create mode 100644 PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs create mode 100644 PowerSync/PowerSync.Common/Client/Sync/Stream/CheckpointState.cs create mode 100644 PowerSync/PowerSync.Common/Utils/BroadcastChannel.cs rename PowerSync/PowerSync.Common/Utils/Converters/{LongToStringConverter.cs => StringLongConverter.cs} (77%) diff --git a/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs b/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs index 4488e6d1..a2afc2c1 100644 --- a/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs +++ b/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs @@ -42,7 +42,7 @@ public interface IPowerSyncBackendConnector public interface ICustomCheckpointRequestConnector : IPowerSyncBackendConnector { /// - /// TODO + /// Posts a client-generated checkpoint request to the backend and returns the effective checkpoint request state. /// - Task PostCheckpointRequest(string clientId, long requestId); + Task PostCheckpointRequest(string clientId, string requestId); } diff --git a/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs b/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs index 36faa46a..083c9e7a 100644 --- a/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs +++ b/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs @@ -207,7 +207,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,6 +226,9 @@ public PowerSyncDatabase(PowerSyncDatabaseOptions options) await WaitForReady(); await connector.UploadData(this); }, + PostCheckpointRequest = (connector is ICustomCheckpointRequestConnector c) + ? (string clientId, string requestId) => c.PostCheckpointRequest(clientId, requestId) + : (_, _) => null, RetryDelayMs = options.RetryDelayMs, Subscriptions = options.Subscriptions, CrudUploadThrottleMs = options.CrudUploadThrottleMs, diff --git a/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs index 151dace0..48faa520 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; @@ -18,7 +19,6 @@ public static class PowerSyncControlCommand public const string NOTIFY_TOKEN_REFRESHED = "refreshed_token"; 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"; /// @@ -137,8 +137,9 @@ public interface IBucketStorageAdapter : ICloseable Task UpdateLocalTarget(Func> callback); Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null); - // TODO type safety - Task ReadCheckpointRequestId(string variant, string? payload = null); + + // TODO Return int64 from this in future release + Task ReadOrUpdateCheckpoint(string variant, string? update = null); /// /// Get a unique client ID. diff --git a/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs b/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs index da3e52dd..daeafadb 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,36 @@ 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? payload = null) + => db.WriteTransaction(tx => ReadOrUpdateCheckpoint(tx, variant, payload)); + + /// + /// 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 accepts ITransaction + private static Task TargetCheckpointRequestId(ITransaction tx, string? update = null) + => ReadOrUpdateCheckpoint(tx, "target", update); + + // These are called from external functions, therefore create transaction + internal Task CurrentCheckpointRequestId(string? update = null) + => db.WriteTransaction(tx => ReadOrUpdateCheckpoint(tx, "current", update)); + + internal Task NextCheckpointRequestId(string? update = null) + => db.WriteTransaction(tx => ReadOrUpdateCheckpoint(tx, "next", update))!; + + internal Task SeedCheckpointRequestId(string? update = null) + => db.WriteTransaction(tx => ReadOrUpdateCheckpoint(tx, "seed", update)); + private record ResultResult(object result); public class ResultDetail @@ -154,6 +171,7 @@ public async Task UpdateLocalTarget(Func> callback) return true; }); } + public Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null) { return db.WriteTransaction(async tx => @@ -198,16 +216,6 @@ await TargetCheckpointRequestId( ); } - public async Task PostCheckpointRequestId(string variant, long? payload = null) - { - string rawResponse = await Control($"{variant}_checkpoint_request_id", payload); - if (long.TryParse(rawResponse, out var response)) - { - return response; - } - throw new Exception($"Expected {variant}_checkpoint_request_id to return value of type long."); - } - public async Task NextCrudItem() { var next = await db.GetOptional("SELECT * FROM ps_crud ORDER BY id ASC LIMIT 1"); diff --git a/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs b/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs new file mode 100644 index 00000000..3c129c80 --- /dev/null +++ b/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs @@ -0,0 +1,22 @@ +namespace PowerSync.Common.Client.Sync; + +/// An exception related to checkpoint requests. +public class CheckpointRequestException : Exception +{ + private CheckpointRequestException(string message) : base(message) { } + + /// "The PowerSync service does not support checkpoint requests. Update to PowerSync service version 1.24.0 or later to use this API." + internal static readonly CheckpointRequestException InstanceNotSupported = new( + "The PowerSync service does not support checkpoint requests. Update to PowerSync service version 1.24.0 or later to use this API." + ); + + /// "Cannot request checkpoints, sync client is disconnected" + internal static readonly CheckpointRequestException Disconnected = new( + "Cannot request checkpoints, sync client is disconnected" + ); + + /// "Connected with legacy checkpoint mode, cannot request checkpoints" + internal static readonly CheckpointRequestException Disabled = new( + "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..2dddf1cf --- /dev/null +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/CheckpointState.cs @@ -0,0 +1,120 @@ +// TODO CheckpointStateTests.cs +using PowerSync.Common.Utils; + +namespace PowerSync.Common.Client.Sync.Stream; + +internal class CheckpointStateSignals +{ + private CheckpointState _state = new CheckpointState.Pending(); + private readonly BroadcastChannel _stateBroadcaster = new(); + private TaskCompletionSource _waitingForCheckpointsReady = new(); + + // -- Check behaviour + private void UpdateState(CheckpointState state) + { + // TODO Run this asynchronously in another Task? + _state = state; + _stateBroadcaster.Broadcast(state); + } + + /// + /// Marks the current download iteration as ended, blocking new checkpoint requests until the + /// seed was performed in the next iteration. + /// + public void DownloadIterationEnded() + { + _waitingForCheckpointsReady = new(); + UpdateState(new CheckpointState.Pending()); + } + + /// + /// Marks the sync client as disconnected, failing all outstanding checkpoint + /// requests and preventing new ones. + /// + public void Disconnect() + { + UpdateState(new CheckpointState.Disconnected()); + } + + /// Waits for a waiter wanting torequest a checkpoint. + /// + /// As the waiter is blocked for a seed run we start in the download + /// iteration we use this to wake up the download iteration if it's currently + /// paused. + public Task WaitForCheckpointWaiter() => _waitingForCheckpointsReady.Task; + + public void MarkCheckpointsReady() + { + UpdateState(new CheckpointState.Ready()); + } + + public void MarkCheckpointsFailed(Exception ex) + { + UpdateState(new CheckpointState.Error(ex)); + } + + public Task WaitForCheckpointRequestsReady(CancellationToken signal, bool wakeDownloadLoop = true) + { + var tcs = new TaskCompletionSource(); + var reader = _stateBroadcaster.Subscribe(out var subscriberId); + + void UnsubscribeReader() => _stateBroadcaster.Unsubscribe(subscriberId); + + // Resolves the promise from the current state if possible, returning true if it was + bool HandleState(CheckpointState state) + { + switch (state) + { + case CheckpointState.Disconnected: + tcs.TrySetException(CheckpointRequestException.Disconnected); + UnsubscribeReader(); + return true; + + case CheckpointState.Ready: + tcs.TrySetResult(true); + UnsubscribeReader(); + return true; + + case CheckpointState.Error e: + tcs.TrySetException(e.Exception); + UnsubscribeReader(); + return true; + + case CheckpointState.Pending: + if (wakeDownloadLoop) + { + _waitingForCheckpointsReady.TrySetResult(true); + } + return false; + } + return false; + } + + // Listen to state changes until task resolves + var cts = CancellationTokenSource.CreateLinkedTokenSource(signal); + _ = Task.Run(async () => + { + while (reader.TryRead(out var state)) + { + if (HandleState(state)) + { + cts.Cancel(); + } + } + }, cts.Token); + HandleState(_state); + + return tcs.Task; + // TODO CHECK IF THIS WORKS + } +} + +internal record CheckpointState +{ + private CheckpointState() { } + + public sealed record Pending : CheckpointState; + public sealed record Disconnected : CheckpointState; + public sealed record Ready : CheckpointState; + public sealed record Error(Exception Exception) : CheckpointState; +} diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs index 128f62ff..d3c44bc6 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs @@ -135,8 +135,7 @@ public class CoreSyncStatus public List Streams { get; set; } = []; [JsonProperty("internal_last_applied_checkpoint_request_id")] - [JsonConverter(typeof(LongToStringConverter))] - public long? LastAppliedCheckpointRequestId { get; set; } + 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 6b955a3d..0cb7edf9 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/Remote.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/Remote.cs @@ -117,11 +117,11 @@ public virtual async Task FetchJson(string path, HttpMethod? method = null if (!response.IsSuccessStatusCode) { - var errorMessage = await response.Content.ReadAsStringAsync(ct); + var errorMessage = await response.Content.ReadAsStringAsync(); throw new HttpRequestException($"Received {response.StatusCode} - {response.ReasonPhrase} when getting from {path}: {errorMessage}"); } - var responseData = await response.Content.ReadAsStringAsync(ct); + var responseData = await response.Content.ReadAsStringAsync(); return JsonConvert.DeserializeObject(responseData)!; } @@ -145,11 +145,11 @@ public virtual async Task PostStreamRaw(SyncStreamOptions options) if (!response.IsSuccessStatusCode) { - var errorText = await response.Content.ReadAsStringAsync(options.CancellationToken); + var errorText = await response.Content.ReadAsStringAsync(); throw new HttpRequestException($"HTTP {response.StatusCode}: {errorText}"); } - return await response.Content.ReadAsStreamAsync(options.CancellationToken); + return await response.Content.ReadAsStreamAsync(); } private async Task BuildRequest(HttpMethod method, string path, object? data = null, Dictionary? additionalHeaders = null) diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs index 2396be50..863e7bd5 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs @@ -50,6 +50,8 @@ public class StreamingSyncImplementationOptions : AdditionalConnectionOptions public Func UploadCrud { get; init; } = null!; + public Func?> PostCheckpointRequest = null!; + public Remote Remote { get; init; } = null!; public ILogger? Logger { get; init; } @@ -175,6 +177,8 @@ public class StreamingSyncImplementation : ICloseable private CancellationTokenSource? crudUpdateCts; private Task? crudUpdateTask; + private readonly CheckpointStateSignals _checkpointState = new(); + private readonly ILogger logger; private SubscribedStream[] activeStreams; @@ -322,14 +326,12 @@ public async Task Disconnect() UpdateSyncStatus(new SyncStatusOptions { Connected = false, Connecting = false }); } - private async Task RequestNextCheckpointFromService(CancellationToken signal) + private async Task RequestNextCheckpointFromService(CancellationToken signal) { - // TODO CheckpointState manager - await Checkpoints.WaitForCheckpointRequestsReady(signal); + await _checkpointState.WaitForCheckpointRequestsReady(signal); // TODO implement on adapter - // TODO type safety - var nextCheckpointRequestId = await Options.Adapter.ReadCheckpointRequestId("next"); + var nextCheckpointRequestId = await Options.Adapter.ReadOrUpdateCheckpoint("next"); var clientId = await Options.Adapter.GetClientId(); return await RequestCheckpointFromService(signal, new() { @@ -338,12 +340,13 @@ private async Task RequestNextCheckpointFromService(CancellationToken sign }); } - private async Task RequestCheckpointFromService(CancellationToken signal, CheckpointRequestPayload request) + private async Task RequestCheckpointFromService(CancellationToken signal, CheckpointRequestPayload request) { // First, check if we can use a custom checkpoint request implementation. - // TODO add and default implement PostCheckpointRequest - var customResponse = await Options.PostCheckpointRequest(request.ClientId, request.CheckpointRequestId); - if (customResponse != null) return customResponse; + if (Options.PostCheckpointRequest != null) + { + return await Options.PostCheckpointRequest(request.ClientId, request.CheckpointRequestId); + } var status = await Options.Remote.FetchJson( path: "/sync/checkpoint-request", @@ -1000,10 +1003,10 @@ public void UpdateSubscriptions(SubscribedStream[] subscriptions) handleActiveStreamsChange?.Invoke(); } - public record LegacyWriteCheckpointResponseData( + private record LegacyWriteCheckpointResponseData( [property: JsonProperty("write_checkpoint")] string WriteCheckpoint ); - public record LegacyWriteCheckpointApiResponse( + private record LegacyWriteCheckpointApiResponse( [property: JsonProperty("data")] LegacyWriteCheckpointResponseData Data ); } diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs index 18a1b372..4253e155 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs @@ -198,8 +198,7 @@ public class CheckpointRequestPayload public string ClientId { get; set; } [JsonProperty("checkpoint_request_id")] - [JsonConverter(typeof(LongToStringConverter))] - public long CheckpointRequestId { get; set; } + public string CheckpointRequestId { get; set; } } public class CheckpointRequestResponse @@ -211,7 +210,5 @@ public class CheckpointRequestResponse public class CheckpointRequestResponseData { [JsonProperty("checkpoint_request_id")] - // TODO Uncomment and implement (copy from temp branch) - // [JsonConverter(typeof(LongToStringConverter))] - public long CheckpointRequestId { get; set; } + public string CheckpointRequestId { get; set; } } 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/LongToStringConverter.cs b/PowerSync/PowerSync.Common/Utils/Converters/StringLongConverter.cs similarity index 77% rename from PowerSync/PowerSync.Common/Utils/Converters/LongToStringConverter.cs rename to PowerSync/PowerSync.Common/Utils/Converters/StringLongConverter.cs index 92eabf86..61c7cbdf 100644 --- a/PowerSync/PowerSync.Common/Utils/Converters/LongToStringConverter.cs +++ b/PowerSync/PowerSync.Common/Utils/Converters/StringLongConverter.cs @@ -3,11 +3,15 @@ namespace PowerSync.Common.Utils.Converters; /// -/// Converts a long to a string representation when converting JSON values. Used +/// 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 LongToStringConverter : JsonConverter +internal class StringLongConverter : JsonConverter { public override bool CanConvert(Type objectType) { diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs index 5b390cc3..a5ab3597 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs @@ -98,8 +98,8 @@ SemaphoreSlim signal 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/Utils/Sync/MockSyncService.cs b/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs index 1d84c77b..c837cb57 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs @@ -212,8 +212,8 @@ public override Task FetchJson(string path, HttpMethod? method = null, obj { if (path.Contains("checkpoint2.json")) { - var response = (T)(object)new StreamingSyncImplementation.ApiResponse( - new StreamingSyncImplementation.ResponseData("1") + var response = (T)(object)new StreamingSyncImplementation.LegacyWriteCheckpointApiResponse( + new StreamingSyncImplementation.LegacyWriteCheckpointResponseData("1") ); return Task.FromResult(response); } @@ -242,6 +242,14 @@ 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) + => _postCheckpointRequest(clientId, requestId); +} + public record LogRecord(LogLevel LogLevel, string CategoryName, string Message, Exception? Exception); public class ListLogger(string categoryName, ConcurrentQueue drain) : ILogger @@ -254,7 +262,7 @@ public void Log(LogLevel logLevel, EventId eventId, TState state, Except _drain.Enqueue(new(logLevel, _categoryName, formatter(state, exception), exception)); } - public IDisposable BeginScope(TState state) => null; + public IDisposable BeginScope(TState state) => null!; public bool IsEnabled(LogLevel logLevel) => true; } From 9ea7be93b64c6246fec4fb7dedad6c17fb3f20e4 Mon Sep 17 00:00:00 2001 From: LucDeCaf Date: Tue, 1 Sep 2026 14:29:33 +0200 Subject: [PATCH 5/9] Refactor StreamingSyncImplementation, rewrite CheckpointStateSignals, tests, more wiring --- .../Client/PowerSyncDatabase.cs | 19 +- .../Sync/Bucket/BucketStorageAdapter.cs | 1 - .../Client/Sync/Bucket/SqliteBucketStorage.cs | 12 +- .../Client/Sync/Stream/CheckpointState.cs | 223 +++++++---- .../Stream/StreamingSyncImplementation.cs | 356 +++++++++++++++--- .../PowerSync.Common/PowerSync.Common.csproj | 1 + .../Utils/BroadcastChannel.cs | 47 --- .../Client/Sync/CheckpointRequestsTests.cs | 238 +++++++++++- .../Sync/SyncIterationControlFlowTests.cs | 1 + .../PowerSync.Common.Tests.csproj | 1 + .../Utils/Sync/MockSyncService.cs | 92 +++-- 11 files changed, 758 insertions(+), 233 deletions(-) delete mode 100644 PowerSync/PowerSync.Common/Utils/BroadcastChannel.cs diff --git a/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs b/PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs index 083c9e7a..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); @@ -226,12 +234,17 @@ public PowerSyncDatabase(PowerSyncDatabaseOptions options) await WaitForReady(); await connector.UploadData(this); }, - PostCheckpointRequest = (connector is ICustomCheckpointRequestConnector c) - ? (string clientId, string requestId) => c.PostCheckpointRequest(clientId, requestId) - : (_, _) => null, + 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 }); diff --git a/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs index 48faa520..227380ed 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs @@ -19,7 +19,6 @@ public static class PowerSyncControlCommand public const string NOTIFY_TOKEN_REFRESHED = "refreshed_token"; 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. diff --git a/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs b/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs index daeafadb..216235f4 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs @@ -76,7 +76,7 @@ public async Task GetClientId() /// /// Reads or updates the stored checkpoint request id using the given transaction. /// - public static Task ReadOrUpdateCheckpoint(ITransaction tx, string variant, string? payload = null) + private static Task ReadOrUpdateCheckpoint(ITransaction tx, string variant, string? payload = null) { // TODO Return 64-bit integer in later release. return tx.Get( @@ -88,16 +88,6 @@ public async Task GetClientId() private static Task TargetCheckpointRequestId(ITransaction tx, string? update = null) => ReadOrUpdateCheckpoint(tx, "target", update); - // These are called from external functions, therefore create transaction - internal Task CurrentCheckpointRequestId(string? update = null) - => db.WriteTransaction(tx => ReadOrUpdateCheckpoint(tx, "current", update)); - - internal Task NextCheckpointRequestId(string? update = null) - => db.WriteTransaction(tx => ReadOrUpdateCheckpoint(tx, "next", update))!; - - internal Task SeedCheckpointRequestId(string? update = null) - => db.WriteTransaction(tx => ReadOrUpdateCheckpoint(tx, "seed", update)); - private record ResultResult(object result); public class ResultDetail diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/CheckpointState.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/CheckpointState.cs index 2dddf1cf..9411fa2f 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/CheckpointState.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/CheckpointState.cs @@ -1,120 +1,189 @@ -// TODO CheckpointStateTests.cs -using PowerSync.Common.Utils; - namespace PowerSync.Common.Client.Sync.Stream; -internal class CheckpointStateSignals +using System.Runtime.ExceptionServices; +using System.Threading.Channels; + +/// +/// 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 TaskCompletionSource _waitingForCheckpointsReady = new(); + private readonly object gate = new(); - // -- Check behaviour - private void UpdateState(CheckpointState state) - { - // TODO Run this asynchronously in another Task? - _state = state; - _stateBroadcaster.Broadcast(state); - } + private CheckpointState state = new CheckpointState.Pending(); + + /// + /// One entry per caller currently blocked in . + /// Completed by so every waiter observes each transition. + /// + private readonly List> stateWaiters = []; + + /// + /// Signalled when a caller starts waiting for checkpoint requests to become available. Used to + /// resume a download iteration that is currently sitting in its retry delay. + /// + private Channel checkpointWaiterArrived = CreateNotifier(); /// /// Marks the current download iteration as ended, blocking new checkpoint requests until the - /// seed was performed in the next iteration. + /// seed performed by the next iteration completes. /// public void DownloadIterationEnded() { - _waitingForCheckpointsReady = new(); - UpdateState(new CheckpointState.Pending()); + lock (gate) + { + // Waiters arriving after this should be able to resume the next download iteration. + checkpointWaiterArrived = CreateNotifier(); + SetState(new CheckpointState.Pending()); + } } /// - /// Marks the sync client as disconnected, failing all outstanding checkpoint - /// requests and preventing new ones. + /// Marks the sync client as disconnected, failing all outstanding checkpoint requests and + /// preventing new ones. /// - public void Disconnect() + public void Disconnected() { - UpdateState(new CheckpointState.Disconnected()); + lock (gate) + { + SetState(new CheckpointState.Disconnected()); + } } - /// Waits for a waiter wanting torequest a checkpoint. - /// - /// As the waiter is blocked for a seed run we start in the download - /// iteration we use this to wake up the download iteration if it's currently - /// paused. - public Task WaitForCheckpointWaiter() => _waitingForCheckpointsReady.Task; - - public void MarkCheckpointsReady() + /// + /// 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) { - UpdateState(new CheckpointState.Ready()); - } + try + { + await seed(); + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception ex) + { + lock (gate) + { + SetState(new CheckpointState.Failed(ex)); + } + throw; + } - public void MarkCheckpointsFailed(Exception ex) - { - UpdateState(new CheckpointState.Error(ex)); + lock (gate) + { + SetState(new CheckpointState.Ready()); + } } - public Task WaitForCheckpointRequestsReady(CancellationToken signal, bool wakeDownloadLoop = true) + /// + /// 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) { - var tcs = new TaskCompletionSource(); - var reader = _stateBroadcaster.Subscribe(out var subscriberId); + ChannelReader reader; + lock (gate) + { + reader = checkpointWaiterArrived.Reader; + } - void UnsubscribeReader() => _stateBroadcaster.Unsubscribe(subscriberId); + await reader.ReadAsync(signal); + } - // Resolves the promise from the current state if possible, returning true if it was - bool HandleState(CheckpointState state) + /// + /// Waits until a download iteration is active and has seeded the checkpoint state, meaning that + /// checkpoint request ids can safely be allocated. + /// + /// Cancelled when the sync client disconnects. + /// + /// Whether a paused download loop should be resumed to seed the state. Callers that only want to + /// piggyback on an iteration someone else needs should pass false. + /// + /// + /// Thrown when the client is disconnected, or when seeding the checkpoint state failed. + /// + public async Task WaitForCheckpointRequestsReady(CancellationToken signal, bool wakeDownloadLoop = true) + { + while (true) { - switch (state) + signal.ThrowIfCancellationRequested(); + + TaskCompletionSource waiter; + lock (gate) { - case CheckpointState.Disconnected: - tcs.TrySetException(CheckpointRequestException.Disconnected); - UnsubscribeReader(); - return true; - - case CheckpointState.Ready: - tcs.TrySetResult(true); - UnsubscribeReader(); - return true; - - case CheckpointState.Error e: - tcs.TrySetException(e.Exception); - UnsubscribeReader(); - return true; - - case CheckpointState.Pending: - if (wakeDownloadLoop) - { - _waitingForCheckpointsReady.TrySetResult(true); - } - return false; + switch (state) + { + case CheckpointState.Ready: + return; + case CheckpointState.Disconnected: + throw CheckpointRequestException.Disconnected; + case CheckpointState.Failed failed: + ExceptionDispatchInfo.Capture(failed.Exception).Throw(); + return; + } + + // Pending: wait for the next transition, optionally asking the download loop to start + // an iteration which can seed the state we're waiting for. + waiter = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + stateWaiters.Add(waiter); + + if (wakeDownloadLoop) + { + checkpointWaiterArrived.Writer.TryWrite(true); + } } - return false; - } - // Listen to state changes until task resolves - var cts = CancellationTokenSource.CreateLinkedTokenSource(signal); - _ = Task.Run(async () => - { - while (reader.TryRead(out var state)) + using var registration = signal.Register(() => waiter.TrySetCanceled(signal)); + try + { + await waiter.Task; + } + finally { - if (HandleState(state)) + lock (gate) { - cts.Cancel(); + stateWaiters.Remove(waiter); } } - }, cts.Token); - HandleState(_state); + } + } + + private void SetState(CheckpointState next) + { + state = next; - return tcs.Task; - // TODO CHECK IF THIS WORKS + foreach (var waiter in stateWaiters) + { + waiter.TrySetResult(true); + } } + + /// 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 CheckpointState +internal abstract record CheckpointState { private CheckpointState() { } + /// No iteration has seeded the checkpoint state, requests have to wait. public sealed record Pending : CheckpointState; + + /// The sync client is disconnected, requests cannot be made at all. public sealed record Disconnected : CheckpointState; + + /// The active iteration has seeded its state, requests can be made. public sealed record Ready : CheckpointState; - public sealed record Error(Exception Exception) : CheckpointState; + + /// Seeding the checkpoint state failed. + public sealed record Failed(Exception Exception) : CheckpointState; } diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs index 863e7bd5..dc7b6782 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; @@ -50,11 +51,21 @@ public class StreamingSyncImplementationOptions : AdditionalConnectionOptions public Func UploadCrud { get; init; } = null!; - public Func?> PostCheckpointRequest = 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, CheckpointMode? checkpointMode = null) @@ -172,21 +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(); + 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) @@ -207,26 +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; - }); - }; - } /// @@ -234,7 +232,6 @@ public StreamingSyncImplementation(StreamingSyncImplementationOptions options) /// public bool IsConnected => SyncStatus.Connected; - /// /// The timestamp of the last successful sync. /// @@ -309,31 +306,29 @@ public async Task Disconnect() streamingSyncTask = null; CancellationTokenSource = null; - // Do the same for any pending CRUD uploads - if (crudUploadTask != null) - { - try - { - await crudUploadTask; - } - catch (Exception ex) - { - logger.LogWarning("CRUD upload task failed during disconnect: {Message}", ex.Message); - } - 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); + await checkpointState.WaitForCheckpointRequestsReady(signal); - // TODO implement on adapter - var nextCheckpointRequestId = await Options.Adapter.ReadOrUpdateCheckpoint("next"); + var nextCheckpointRequestId = await Options.Adapter.ReadOrUpdateCheckpoint("next") + ?? throw new InvalidOperationException("The core extension did not return a checkpoint request id."); var clientId = await Options.Adapter.GetClientId(); - return await RequestCheckpointFromService(signal, new() + return await RequestCheckpointFromService(signal, new CheckpointRequestPayload { ClientId = clientId, CheckpointRequestId = nextCheckpointRequestId, @@ -357,13 +352,24 @@ private async Task RequestCheckpointFromService(CancellationToken signal return status.Data.CheckpointRequestId; } - // TODO convert write checkpoint data type to long + /// + /// 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.ReadOrUpdateCheckpoint("seed", 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; } @@ -375,6 +381,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 () => { @@ -387,7 +416,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(); @@ -413,7 +442,7 @@ protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectio try { - if (signal.Value.IsCancellationRequested) + if (signal.IsCancellationRequested) { break; } @@ -460,7 +489,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(); @@ -475,7 +504,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); } } } @@ -488,6 +519,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), + Options.TimeProvider.Delay(TimeSpan.FromMilliseconds(throttleMs), signal) + ); + + 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.ReadOrUpdateCheckpoint("current"); + + // 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.ReadOrUpdateCheckpoint("current")) + { + 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; } @@ -499,6 +643,12 @@ 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) @@ -531,6 +681,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(); }); @@ -659,6 +812,7 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) } break; case UpdateSyncStatus syncStatus: + lastAppliedCheckpointRequestId = syncStatus.Status.LastAppliedCheckpointRequestId; UpdateSyncStatus(CoreInstructionHelpers.CoreStatusToSyncStatusOptions(syncStatus.Status)); break; case FetchCredentials fetchCredentials: @@ -705,7 +859,7 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) active_streams = activeStreams, include_defaults = resolvedOptions.IncludeDefaultStreams, app_metadata = resolvedOptions.AppMetadata, - checkpoint_mode = resolvedOptions.CheckpointMode == CheckpointMode.Legacy ? "legacy" : "requests", + checkpoint_mode = resolvedOptions.CheckpointMode is CheckpointMode.Requests ? "requests" : "legacy", }; StreamingSyncRequest? establishRequest = null; @@ -756,6 +910,37 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) }); } }; + + if (establish.CheckpointRequest is { } seedRequest) + { + // Reconcile checkpoint state on every started download iteration to: + // 1. Align service and client checkpoint ids, allowing both parties to + // safely forget old checkpoints. + // 2. If the user id changes between connections, agree on the highest + // checkpoint request between the old and new user to make sure we'll + // receive that checkpoint eventually. + // This runs concurrently so that sync lines are processed while it's pending. + seedingCheckpointState = Task.Run(async () => + { + try + { + await checkpointState.MarkCheckpointsReady( + () => SeedCheckpointRequestState(nestedCts.Token, seedRequest)); + } + catch (OperationCanceledException) + { + // Tearing down, the next iteration will seed again. + } + 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) { @@ -779,6 +964,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)) { @@ -839,6 +1029,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(); } @@ -853,9 +1053,8 @@ public void Close() Events.Close(); } - protected async Task InternalUploadAllCrud() + protected async Task InternalUploadAllCrud(CancellationToken signal, PowerSyncConnectionOptions options) { - await locks.ObtainLock(new LockOptions { Type = LockType.CRUD, @@ -863,16 +1062,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( @@ -897,10 +1096,27 @@ await locks.ObtainLock(new LockOptions else { // Uploading is completed - await Options.Adapter.UpdateLocalTarget(GetLegacyWriteCheckpoint); + 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; @@ -913,7 +1129,7 @@ await locks.ObtainLock(new LockOptions } }); - await DelayRetry(); + await DelayRetry(signal, options.RetryDelayMs ?? DEFAULT_RETRY_DELAY_MS); if (!IsConnected) { @@ -989,12 +1205,29 @@ protected void UpdateSyncStatus(SyncStatusOptions options, UpdateSyncStatusOptio } } - private async Task DelayRetry() + /// + /// 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 (!resumeOnCheckpointRequest) { - await Task.Delay(Options.RetryDelayMs.Value); + await Options.TimeProvider.Delay(TimeSpan.FromMilliseconds(delay), signal); + return; } + + using var nestedCts = CancellationTokenSource.CreateLinkedTokenSource(signal); + await Task.WhenAny( + checkpointState.WaitForCheckpointWaiter(nestedCts.Token), + Options.TimeProvider.Delay(TimeSpan.FromMilliseconds(delay), nestedCts.Token)); + + // Ends the loser. Without this an abandoned checkpoint waiter would consume the signal that + // should have woken the next delay. + nestedCts.Cancel(); + + signal.ThrowIfCancellationRequested(); } public void UpdateSubscriptions(SubscribedStream[] subscriptions) @@ -1003,10 +1236,13 @@ public void UpdateSubscriptions(SubscribedStream[] subscriptions) handleActiveStreamsChange?.Invoke(); } - private record LegacyWriteCheckpointResponseData( + /// 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 ); - private record LegacyWriteCheckpointApiResponse( + internal record LegacyWriteCheckpointApiResponse( [property: JsonProperty("data")] LegacyWriteCheckpointResponseData Data ); } diff --git a/PowerSync/PowerSync.Common/PowerSync.Common.csproj b/PowerSync/PowerSync.Common/PowerSync.Common.csproj index 9f4e1cc8..06c02423 100644 --- a/PowerSync/PowerSync.Common/PowerSync.Common.csproj +++ b/PowerSync/PowerSync.Common/PowerSync.Common.csproj @@ -30,6 +30,7 @@ + diff --git a/PowerSync/PowerSync.Common/Utils/BroadcastChannel.cs b/PowerSync/PowerSync.Common/Utils/BroadcastChannel.cs deleted file mode 100644 index 92a4bc97..00000000 --- a/PowerSync/PowerSync.Common/Utils/BroadcastChannel.cs +++ /dev/null @@ -1,47 +0,0 @@ -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/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs index 3d84440f..6694cba4 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs @@ -1,9 +1,11 @@ -using PowerSync.Common.Tests.Utils; -using PowerSync.Common.Tests.Utils.Sync; +using Microsoft.Extensions.Time.Testing; + using PowerSync.Common.Client; using PowerSync.Common.Client.Connection; -using PowerSync.Common.Client.Sync.Stream; 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; @@ -15,6 +17,9 @@ 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(); @@ -30,7 +35,7 @@ public async Task DisposeAsync() DatabaseUtils.CleanDb(_db.Database.Name); } - [Fact] + [Fact(Timeout = 15000)] public async Task CheckpointRequests_WarnsCustomConnectorWithoutRequestsEnabled() { await _db.Connect(new CheckpointRequestConnector()); @@ -40,11 +45,12 @@ public async Task CheckpointRequests_WarnsCustomConnectorWithoutRequestsEnabled( Assert.Contains("implements ICustomCheckpointRequestConnector, but Connect() was called without checkpoint requests enabled.", logs[0].Message); } - [Fact] + [Fact(Timeout = 15000)] public async Task CheckpointRequests_RequestsCheckpointsForUpdates() { - await _db.Connect(new CheckpointRequestConnector(), new(checkpointMode: new CheckpointMode.Requests())); + 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"]); @@ -54,7 +60,7 @@ public async Task CheckpointRequests_RequestsCheckpointsForUpdates() Assert.Single(watched.Current); Assert.Equal("local write", watched.Current[0].name); - // The local write should eventually be uploaded. + // The local write should eventually be uploaded, which requests a checkpoint. await TestUtils.WaitForAsync(() => _syncService.CheckpointRequests.Count == 2); _syncService.PushLine(new StreamingSyncCheckpoint @@ -62,7 +68,7 @@ public async Task CheckpointRequests_RequestsCheckpointsForUpdates() Checkpoint = new() { LastOpId = "1", - Buckets = [new() { Bucket = "a", Count = 1, Checksum = 0, Priority = 3 }], + Buckets = [MockDataFactory.Bucket("a", 1, subscriptions: Array.Empty())], WriteCheckpoint = _syncService.LastWriteCheckpoint.ToString(), } }); @@ -88,12 +94,224 @@ public async Task CheckpointRequests_RequestsCheckpointsForUpdates() await watched.MoveNextAsync(); Assert.Empty(watched.Current); } - private record NameResult(string name); + + [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, long requestId) + public Task PostCheckpointRequest(string clientId, string requestId) { return Task.FromResult(requestId); } 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/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 c837cb57..2238cc11 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Utils/Sync/MockSyncService.cs @@ -25,10 +25,59 @@ public class MockSyncService : EventStream private readonly ListLoggerProvider _listLoggerProvider = new(); public IReadOnlyList Logs => _listLoggerProvider.Logs; - public long LastWriteCheckpoint { get; set; } = 0; + 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; - private readonly List _checkpointRequests = []; - public IReadOnlyList CheckpointRequests => _checkpointRequests; + /// + /// 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) { @@ -40,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(); @@ -51,6 +100,7 @@ public PowerSyncDatabase CreateDatabase(string? dbFilename = null) Database = new SQLOpenOptions { DbFilename = dbFilename }, Schema = TestSchemaTodoList.AppSchema, RemoteFactory = _ => mockRemote, + TimeProvider = timeProvider, Logger = CreateLogger() }); } @@ -162,7 +212,6 @@ public MockRemote( this.connectedListeners = connectedListeners; } - // TODO This should be able to parse and handle /sync/stream AND /sync/checkpoint_request (or whatever the URL is) public override Task PostStreamRaw(SyncStreamOptions options) { if (options.Path.EndsWith("/sync/stream")) @@ -194,31 +243,26 @@ public override Task PostStreamRaw(SyncStreamOptions options) return Task.FromResult(pipe.Reader.AsStream()); } - else if (options.Path.Contains("/sync/checkpoint-request")) - { - // TODO - throw new Exception("Not implemented"); - } - else if (options.Path.Contains("/write-checkpoint2.json")) - { - // TODO - throw new Exception("Not implemented"); - } - throw new Exception("Not implemented"); + throw new InvalidOperationException($"MockRemote received an unexpected stream request: {options.Path}"); } - public override Task FetchJson(string path, HttpMethod? method = null, object? data = null, Dictionary? headers = null, CancellationToken ct = default) + public override async Task FetchJson(string path, HttpMethod? method = null, object? data = null, Dictionary? headers = null, CancellationToken ct = default) { - if (path.Contains("checkpoint2.json")) + if (path.Contains("/sync/checkpoint-request")) + { + var response = await syncService.HandleCheckpointRequest((CheckpointRequestPayload)data!); + return (T)(object)response; + } + + if (path.Contains("write-checkpoint2.json")) { - var response = (T)(object)new StreamingSyncImplementation.LegacyWriteCheckpointApiResponse( + return (T)(object)new StreamingSyncImplementation.LegacyWriteCheckpointApiResponse( new StreamingSyncImplementation.LegacyWriteCheckpointResponseData("1") ); - return Task.FromResult(response); } - throw new InvalidOperationException("Not implemented"); + throw new InvalidOperationException($"MockRemote received an unexpected request: {path}"); } } @@ -242,11 +286,11 @@ public async Task UploadData(IPowerSyncDatabase database) } } -public class TestCustomCheckpointsConnector(Func> postCheckpointRequest) : TestConnector(), ICustomCheckpointRequestConnector +public class TestCustomCheckpointsConnector(Func> postCheckpointRequest) : TestConnector, ICustomCheckpointRequestConnector { - private readonly Func> _postCheckpointRequest = postCheckpointRequest; + private readonly Func> _postCheckpointRequest = postCheckpointRequest; - public Task PostCheckpointRequest(string clientId, long requestId) + public Task PostCheckpointRequest(string clientId, string requestId) => _postCheckpointRequest(clientId, requestId); } From f9dc1da3b544436c8566058eac96bb986629acfd Mon Sep 17 00:00:00 2001 From: LucDeCaf Date: Wed, 2 Sep 2026 09:19:27 +0200 Subject: [PATCH 6/9] don't throw in DelayRetry --- .../Stream/StreamingSyncImplementation.cs | 42 ++++++++++++++----- 1 file changed, 31 insertions(+), 11 deletions(-) diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs index dc7b6782..38a964e2 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs @@ -298,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 @@ -534,7 +538,7 @@ protected async Task CrudUploadLoop(CancellationToken signal, PowerSyncConnectio // Start the initial CRUD upload on connect. Then, keep polling until we're done. await Task.WhenAll( InternalUploadAllCrud(signal, options), - Options.TimeProvider.Delay(TimeSpan.FromMilliseconds(throttleMs), signal) + DelayRetry(signal, throttleMs) ); await crudUploadRequested.Reader.ReadAsync(signal); @@ -979,7 +983,7 @@ await checkpointState.MarkCheckpointsReady( if (instruction is CloseSyncStream closeSyncStream) { hideDisconnectOnRestart = closeSyncStream.HideDisconnect; - logger.LogWarning("Closing stream"); + logger.LogDebug("Closing stream"); close = true; break; } @@ -1205,6 +1209,10 @@ protected void UpdateSyncStatus(SyncStatusOptions options, UpdateSyncStatusOptio } } + /// + /// 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 @@ -1212,22 +1220,34 @@ protected void UpdateSyncStatus(SyncStatusOptions options, UpdateSyncStatusOptio /// private async Task DelayRetry(CancellationToken signal, int delay, bool resumeOnCheckpointRequest = false) { - if (!resumeOnCheckpointRequest) + if (signal.IsCancellationRequested) { - await Options.TimeProvider.Delay(TimeSpan.FromMilliseconds(delay), signal); return; } using var nestedCts = CancellationTokenSource.CreateLinkedTokenSource(signal); - await Task.WhenAny( - checkpointState.WaitForCheckpointWaiter(nestedCts.Token), - Options.TimeProvider.Delay(TimeSpan.FromMilliseconds(delay), nestedCts.Token)); + var timeout = Options.TimeProvider.Delay(TimeSpan.FromMilliseconds(delay), nestedCts.Token); - // Ends the loser. Without this an abandoned checkpoint waiter would consume the signal that - // should have woken the next delay. - nestedCts.Cancel(); + 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. + } + } - signal.ThrowIfCancellationRequested(); + // 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) From bc58b60825567aeb17f00b9afed1d92f019d7af1 Mon Sep 17 00:00:00 2001 From: LucDeCaf Date: Wed, 2 Sep 2026 09:30:28 +0200 Subject: [PATCH 7/9] reduce test warnings --- .../Client/Sync/CheckpointRequestsTests.cs | 2 +- .../PowerSync.Common.Tests/Client/Sync/SyncStreamsTests.cs | 4 ++++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs index 6694cba4..9fb250f9 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/CheckpointRequestsTests.cs @@ -74,7 +74,7 @@ public async Task CheckpointRequests_RequestsCheckpointsForUpdates() }); _syncService.PushLine(new StreamingSyncDataJSON { - Data = new() + Data = new SyncDataBucketJSON { Bucket = "a", Data = [ 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 From 3f3041a2e6ca810f6dbc535c7294ddfc6b8efd23 Mon Sep 17 00:00:00 2001 From: LucDeCaf Date: Wed, 2 Sep 2026 09:57:28 +0200 Subject: [PATCH 8/9] readd Next/Current/SeedCheckpointRequestId methods --- .../Connection/IPowerSyncBackendConnector.cs | 4 ++ .../Sync/Bucket/BucketStorageAdapter.cs | 38 +++++++++++++++++++ .../Client/Sync/Bucket/SqliteBucketStorage.cs | 8 ++-- .../Stream/StreamingSyncImplementation.cs | 8 ++-- 4 files changed, 50 insertions(+), 8 deletions(-) diff --git a/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs b/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs index a2afc2c1..3dc08ccb 100644 --- a/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs +++ b/PowerSync/PowerSync.Common/Client/Connection/IPowerSyncBackendConnector.cs @@ -43,6 +43,10 @@ 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/Sync/Bucket/BucketStorageAdapter.cs b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs index 227380ed..a0c341e8 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs @@ -138,8 +138,46 @@ public interface IBucketStorageAdapter : ICloseable 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); + /// + /// Increments and returns the local checkpoint counter. + /// + Task NextCheckpointRequestId() + => ReadOrUpdateCheckpoint("next")!; + + /// + /// Returns the highest checkpoint request ID that has been requested on this device. + /// + 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. + /// + /// + /// + /// + Task SeedCheckpointRequestId(string serviceResponse) + => ReadOrUpdateCheckpoint("next", serviceResponse)!; + /// /// Get a unique client ID. /// diff --git a/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs b/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs index 216235f4..2e64404f 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Bucket/SqliteBucketStorage.cs @@ -70,13 +70,13 @@ public async Task GetClientId() /// /// Reads or updates the stored checkpoint request id. /// - public Task ReadOrUpdateCheckpoint(string variant, string? payload = null) - => db.WriteTransaction(tx => ReadOrUpdateCheckpoint(tx, variant, payload)); + 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. /// - private static Task ReadOrUpdateCheckpoint(ITransaction tx, string variant, string? payload = null) + public static Task ReadOrUpdateCheckpoint(ITransaction tx, string variant, string? payload = null) { // TODO Return 64-bit integer in later release. return tx.Get( @@ -84,7 +84,7 @@ public async Task GetClientId() [$"{variant}_checkpoint_request_id", payload]); } - // This is called within existing transactions, therefore accepts ITransaction + // 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); diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs index 38a964e2..30b4d9b0 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs @@ -329,7 +329,7 @@ private async Task RequestNextCheckpointFromService(CancellationToken si { await checkpointState.WaitForCheckpointRequestsReady(signal); - var nextCheckpointRequestId = await Options.Adapter.ReadOrUpdateCheckpoint("next") + 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 @@ -363,7 +363,7 @@ private async Task RequestCheckpointFromService(CancellationToken signal private async Task SeedCheckpointRequestState(CancellationToken signal, CheckpointRequestPayload request) { var seed = await RequestCheckpointFromService(signal, request); - await Options.Adapter.ReadOrUpdateCheckpoint("seed", seed); + await Options.Adapter.SeedCheckpointRequestId(seed); } // TODO convert write checkpoint data type to long in a future release @@ -576,13 +576,13 @@ protected async Task RepostUnacknowledgedCheckpointRequests(CancellationToken si // Never wakes the download loop: this only re-posts what another caller requested. await checkpointState.WaitForCheckpointRequestsReady(signal, wakeDownloadLoop: false); - var requestId = await Options.Adapter.ReadOrUpdateCheckpoint("current"); + 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.ReadOrUpdateCheckpoint("current")) + if (requestId != await Options.Adapter.CurrentCheckpointRequestId()) { continue; } From fa5018d93035f648d6f5ad1e32aae141b611dac1 Mon Sep 17 00:00:00 2001 From: LucDeCaf Date: Wed, 2 Sep 2026 15:59:31 +0200 Subject: [PATCH 9/9] Readd BroadcastChannel, dynamic error creation, fix using non-netstandard2.0 runtime features, use manual locking in CheckpointStateSignals --- .../Sync/Bucket/BucketStorageAdapter.cs | 41 +++--- .../Client/Sync/CheckpointRequest.cs | 27 ++-- .../Client/Sync/Stream/CheckpointState.cs | 134 +++++++----------- .../Stream/StreamingSyncImplementation.cs | 13 +- .../Utils/BroadcastChannel.cs | 47 ++++++ 5 files changed, 143 insertions(+), 119 deletions(-) create mode 100644 PowerSync/PowerSync.Common/Utils/BroadcastChannel.cs diff --git a/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs index a0c341e8..3ccb512c 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs @@ -143,17 +143,36 @@ public interface IBucketStorageAdapter : ICloseable /// Task ReadOrUpdateCheckpoint(string variant, string? update = null); + /// + /// Get a unique client ID. + /// + Task GetClientId(); + + /// + /// Invokes the `powersync_control` function for the sync client. + /// + 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. /// - Task NextCheckpointRequestId() - => ReadOrUpdateCheckpoint("next")!; + public static Task NextCheckpointRequestId(this IBucketStorageAdapter adapter) + => adapter.ReadOrUpdateCheckpoint("next")!; /// /// Returns the highest checkpoint request ID that has been requested on this device. /// - Task CurrentCheckpointRequestId() - => ReadOrUpdateCheckpoint("current"); + public static Task CurrentCheckpointRequestId(this IBucketStorageAdapter adapter) + => adapter.ReadOrUpdateCheckpoint("current"); /// /// Seeds the local checkpoint request ID counter using a response from the server. @@ -175,16 +194,6 @@ Task NextCheckpointRequestId() /// /// /// - Task SeedCheckpointRequestId(string serviceResponse) - => ReadOrUpdateCheckpoint("next", serviceResponse)!; - - /// - /// Get a unique client ID. - /// - Task GetClientId(); - - /// - /// Invokes the `powersync_control` function for the sync client. - /// - Task Control(string op, object? payload); + public static Task SeedCheckpointRequestId(this IBucketStorageAdapter adapter, string serviceResponse) + => adapter.ReadOrUpdateCheckpoint("seed", serviceResponse)!; } diff --git a/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs b/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs index 3c129c80..8f99fb46 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/CheckpointRequest.cs @@ -3,20 +3,21 @@ namespace PowerSync.Common.Client.Sync; /// An exception related to checkpoint requests. public class CheckpointRequestException : Exception { - private CheckpointRequestException(string message) : base(message) { } + /// Initializes a new instance of the class. + public CheckpointRequestException() : base() { } - /// "The PowerSync service does not support checkpoint requests. Update to PowerSync service version 1.24.0 or later to use this API." - internal static readonly CheckpointRequestException InstanceNotSupported = new( - "The PowerSync service does not support checkpoint requests. Update to PowerSync service version 1.24.0 or later to use this API." - ); + /// Initializes a new instance of the class with a specified error message. + public CheckpointRequestException(string message) : base(message) { } - /// "Cannot request checkpoints, sync client is disconnected" - internal static readonly CheckpointRequestException Disconnected = new( - "Cannot request checkpoints, sync client is disconnected" - ); + /// 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) { } - /// "Connected with legacy checkpoint mode, cannot request checkpoints" - internal static readonly CheckpointRequestException Disabled = new( - "Connected with legacy checkpoint mode, cannot request checkpoints" - ); + /// 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 index 9411fa2f..16f020d2 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/CheckpointState.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/CheckpointState.cs @@ -3,27 +3,20 @@ 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 readonly object gate = new(); - - private CheckpointState state = new CheckpointState.Pending(); + private CheckpointState _state = new CheckpointState.Pending(); - /// - /// One entry per caller currently blocked in . - /// Completed by so every waiter observes each transition. - /// - private readonly List> stateWaiters = []; + private readonly BroadcastChannel _stateBroadcaster = new(); + private Channel _checkpointWaiterNotifier = CreateNotifier(); - /// - /// Signalled when a caller starts waiting for checkpoint requests to become available. Used to - /// resume a download iteration that is currently sitting in its retry delay. - /// - private Channel checkpointWaiterArrived = CreateNotifier(); + private readonly object _lock = new(); /// /// Marks the current download iteration as ended, blocking new checkpoint requests until the @@ -31,11 +24,11 @@ internal sealed class CheckpointStateSignals /// public void DownloadIterationEnded() { - lock (gate) + lock (_lock) { // Waiters arriving after this should be able to resume the next download iteration. - checkpointWaiterArrived = CreateNotifier(); - SetState(new CheckpointState.Pending()); + _checkpointWaiterNotifier = CreateNotifier(); + UpdateState(new CheckpointState.Pending()); } } @@ -45,9 +38,9 @@ public void DownloadIterationEnded() /// public void Disconnected() { - lock (gate) + lock (_lock) { - SetState(new CheckpointState.Disconnected()); + UpdateState(new CheckpointState.Disconnected()); } } @@ -68,16 +61,16 @@ public async Task MarkCheckpointsReady(Func seed) } catch (Exception ex) { - lock (gate) + lock (_lock) { - SetState(new CheckpointState.Failed(ex)); + UpdateState(new CheckpointState.Failed(ex)); } throw; } - lock (gate) + lock (_lock) { - SetState(new CheckpointState.Ready()); + UpdateState(new CheckpointState.Ready()); } } @@ -90,9 +83,9 @@ public async Task MarkCheckpointsReady(Func seed) public async Task WaitForCheckpointWaiter(CancellationToken signal) { ChannelReader reader; - lock (gate) + lock (_lock) { - reader = checkpointWaiterArrived.Reader; + reader = _checkpointWaiterNotifier.Reader; } await reader.ReadAsync(signal); @@ -102,71 +95,58 @@ public async Task WaitForCheckpointWaiter(CancellationToken signal) /// Waits until a download iteration is active and has seeded the checkpoint state, meaning that /// checkpoint request ids can safely be allocated. /// - /// Cancelled when the sync client disconnects. - /// - /// Whether a paused download loop should be resumed to seed the state. Callers that only want to - /// piggyback on an iteration someone else needs should pass false. - /// - /// - /// Thrown when the client is disconnected, or when seeding the checkpoint state failed. - /// public async Task WaitForCheckpointRequestsReady(CancellationToken signal, bool wakeDownloadLoop = true) { - while (true) + var reader = _stateBroadcaster.Subscribe(out var subscriberId); + try { - signal.ThrowIfCancellationRequested(); - - TaskCompletionSource waiter; - lock (gate) + while (!HandleState(wakeDownloadLoop)) { - switch (state) - { - case CheckpointState.Ready: - return; - case CheckpointState.Disconnected: - throw CheckpointRequestException.Disconnected; - case CheckpointState.Failed failed: - ExceptionDispatchInfo.Capture(failed.Exception).Throw(); - return; - } - - // Pending: wait for the next transition, optionally asking the download loop to start - // an iteration which can seed the state we're waiting for. - waiter = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - stateWaiters.Add(waiter); - - if (wakeDownloadLoop) - { - checkpointWaiterArrived.Writer.TryWrite(true); - } + await reader.ReadAsync(signal); } + } + finally + { + _stateBroadcaster.Unsubscribe(subscriberId); + } + } - using var registration = signal.Register(() => waiter.TrySetCanceled(signal)); - try - { - await waiter.Task; - } - finally + /// + /// Returns true if checkpoint requests are ready and false if we need + /// to keep waiting. + /// + private bool HandleState(bool wakeDownloadLoop) + { + lock (_lock) + { + switch (_state) { - lock (gate) - { - stateWaiters.Remove(waiter); - } + 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 SetState(CheckpointState next) + private void UpdateState(CheckpointState next) { - state = next; - - foreach (var waiter in stateWaiters) - { - waiter.TrySetResult(true); - } + _state = next; + _stateBroadcaster.Broadcast(true); } - /// A conflating single-slot channel: only the fact that a signal arrived matters. + /// 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 }); } @@ -175,15 +155,11 @@ internal abstract record CheckpointState { private CheckpointState() { } - /// No iteration has seeded the checkpoint state, requests have to wait. public sealed record Pending : CheckpointState; - /// The sync client is disconnected, requests cannot be made at all. public sealed record Disconnected : CheckpointState; - /// The active iteration has seeded its state, requests can be made. public sealed record Ready : CheckpointState; - /// Seeding the checkpoint state failed. public sealed record Failed(Exception Exception) : CheckpointState; } diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs index 30b4d9b0..a1dfea0c 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs @@ -917,13 +917,7 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) if (establish.CheckpointRequest is { } seedRequest) { - // Reconcile checkpoint state on every started download iteration to: - // 1. Align service and client checkpoint ids, allowing both parties to - // safely forget old checkpoints. - // 2. If the user id changes between connections, agree on the highest - // checkpoint request between the old and new user to make sure we'll - // receive that checkpoint eventually. - // This runs concurrently so that sync lines are processed while it's pending. + // Run concurrently so that seeding checkpoint state doesn't block sync line processing. seedingCheckpointState = Task.Run(async () => { try @@ -931,10 +925,7 @@ async Task HandleInstruction(NonInterruptingInstruction instruction) await checkpointState.MarkCheckpointsReady( () => SeedCheckpointRequestState(nestedCts.Token, seedRequest)); } - catch (OperationCanceledException) - { - // Tearing down, the next iteration will seed again. - } + catch (OperationCanceledException) { } catch (Exception ex) { // Fail the download iteration if checkpoint requests are broken. 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); + } + } +} +