From 980440669e872032c854d2ee8b3dad7b64de8083 Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Mon, 27 Jul 2026 13:55:54 +0200 Subject: [PATCH 1/3] Simplify control flow in sync iteration. --- PowerSync/PowerSync.Common/CHANGELOG.md | 3 + .../Sync/Bucket/BucketStorageAdapter.cs | 11 + .../Client/Sync/Stream/CoreInstructions.cs | 12 + .../Stream/StreamingSyncImplementation.cs | 224 ++++++++++++------ PowerSync/PowerSync.Maui/CHANGELOG.md | 4 + .../Client/Sync/SyncIterationTeardownTests.cs | 126 ++++++++++ 6 files changed, 306 insertions(+), 74 deletions(-) create mode 100644 Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationTeardownTests.cs diff --git a/PowerSync/PowerSync.Common/CHANGELOG.md b/PowerSync/PowerSync.Common/CHANGELOG.md index 130cbd9..954bfc5 100644 --- a/PowerSync/PowerSync.Common/CHANGELOG.md +++ b/PowerSync/PowerSync.Common/CHANGELOG.md @@ -1,5 +1,8 @@ # PowerSync.Common Changelog +## 0.1.4 (Unreleased) +- Fix "No iteration is active" errors when a sync iteration ends + ## 0.1.3 - **Breaking:** Made `Table.Name` non-nullable (default ""). This change may affect 0% of users, but it is technically a breaking change. diff --git a/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs index c073561..cdeb4f4 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs @@ -18,6 +18,17 @@ 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"; + + /// + /// An `established` or `end` event for response streams. + /// + public const string CONNECTION_STATE = "connection"; +} + +public static class PowerSyncControlConnectionState +{ + public const string ESTABLISHED = "established"; + public const string END = "end"; } public class Checkpoint diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs index 250f38b..1c07c01 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs @@ -11,6 +11,14 @@ namespace PowerSync.Common.Client.Sync.Stream; /// public abstract class Instruction { + /// + /// Whether this instruction starts or stops a sync iteration + /// ( or ). + /// Interrupting instructions are handled by the iteration control loop; all + /// other (non-interrupting) instructions are handled by + /// HandleInstruction. + /// + public virtual bool IsInterrupting => false; public static Instruction[] ParseInstructions(string rawResponse) { @@ -62,6 +70,8 @@ public class LogLine : Instruction public class EstablishSyncStream : Instruction { + public override bool IsInterrupting => true; + [JsonProperty("request")] public StreamingSyncRequest Request { get; set; } = null!; } @@ -175,6 +185,8 @@ public class FetchCredentials : Instruction public class CloseSyncStream : Instruction { + public override bool IsInterrupting => true; + [JsonProperty("hide_disconnect")] public bool HideDisconnect { get; set; } } diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs index 0dc6417..5531f5b 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs @@ -476,66 +476,53 @@ protected async Task StreamingSyncIteration(Cancel protected async Task RustStreamingSyncIteration(CancellationToken? signal, RequiredPowerSyncConnectionOptions resolvedOptions) { - Task? receivingLines = null; - bool hadSyncLine = false; bool hideDisconnectOnRestart = false; + Action? notifyTokenRefreshed = null; - var controlInvocations = (EventStream)null!; + // A failure opening or reading the stream, surfaced from the control loop so it retries. + Exception? streamError = null; var nestedCts = new CancellationTokenSource(); signal?.Register(() => { nestedCts.Cancel(); }); - async Task Connect(EstablishSyncStream instruction) + async Task ReceiveSyncLines(StreamingSyncRequest request, EventStream sink, CancellationToken token) { var syncOptions = new SyncStreamOptions { Path = "/sync/stream", - CancellationToken = nestedCts.Token, - Data = instruction.Request + CancellationToken = token, + Data = request }; - controlInvocations = new EventStream(); + var established = false; try { - _ = Task.Run(async () => - { - await foreach (var line in controlInvocations.ListenAsync(new CancellationToken())) - { - await Control(line.Command, line.Payload); - - // Triggers a local CRUD upload when the first sync line has been received. - // This allows uploading local changes that have been made while offline or disconnected. - if (!hadSyncLine) - { - TriggerCrudUpload(); - hadSyncLine = true; - } - } - }); - var stream = await Options.Remote.PostStreamRaw(syncOptions); using var reader = new StreamReader(stream, Encoding.UTF8); - syncOptions.CancellationToken.Register(() => + token.Register(() => { try { stream?.Close(); } catch { } }); - UpdateSyncStatus(new SyncStatusOptions + // We're connected here, tell core extension + sink.Emit(new EnqueuedCommand { - Connected = true + Command = PowerSyncControlCommand.CONNECTION_STATE, + Payload = PowerSyncControlConnectionState.ESTABLISHED }); + established = true; // Read lines in a cancellation-aware manner. // ReadLineAsync() doesn't support CancellationToken on all .NET versions, // so we use WhenAny to check for cancellation between reads. - while (!syncOptions.CancellationToken.IsCancellationRequested) + while (!token.IsCancellationRequested) { var readTask = reader.ReadLineAsync(); // Create a task that completes when cancellation is requested var cancellationTcs = new TaskCompletionSource(); - using var registration = syncOptions.CancellationToken.Register(() => cancellationTcs.TrySetResult(true)); + using var registration = token.Register(() => cancellationTcs.TrySetResult(true)); var completedTask = await Task.WhenAny(readTask, cancellationTcs.Task); @@ -552,39 +539,53 @@ async Task Connect(EstablishSyncStream instruction) break; } - controlInvocations?.Emit(new EnqueuedCommand + sink.Emit(new EnqueuedCommand { Command = PowerSyncControlCommand.PROCESS_TEXT_LINE, Payload = line }); } } + catch (Exception ex) + { + if (!token.IsCancellationRequested) + { + streamError = ex; + } + } finally { - var activeInstructions = controlInvocations; - controlInvocations = null; - activeInstructions?.Close(); + if (established && !sink.Closed) + { + sink.Emit(new EnqueuedCommand + { + Command = PowerSyncControlCommand.CONNECTION_STATE, + Payload = PowerSyncControlConnectionState.END + }); + } + + sink.Close(); } } async Task Stop() { - await Control(PowerSyncControlCommand.STOP); + foreach (var instruction in await InvokePowerSyncControl(PowerSyncControlCommand.STOP)) + { + // Unconditionally ending the iteration, so interrupting instructions don't apply. + if (instruction.IsInterrupting) + { + continue; + } + await HandleInstruction(instruction); + } } - async Task Control(string op, object? payload = null) + async Task InvokePowerSyncControl(string op, object? payload = null) { var rawResponse = await Options.Adapter.Control(op, payload); logger.LogTrace("powersync_control {op}, {payload}, {rawResponse}", op, payload, rawResponse); - HandleInstructions(Instruction.ParseInstructions(rawResponse)); - } - - async void HandleInstructions(Instruction[] instructions) - { - foreach (var instruction in instructions) - { - await HandleInstruction(instruction); - } + return Instruction.ParseInstructions(rawResponse); } async Task HandleInstruction(Instruction instruction) @@ -608,14 +609,6 @@ async Task HandleInstruction(Instruction instruction) case UpdateSyncStatus syncStatus: UpdateSyncStatus(CoreInstructionHelpers.CoreStatusToSyncStatusOptions(syncStatus.Status)); break; - case EstablishSyncStream establishSyncStream: - if (receivingLines != null) - { - throw new Exception("Unexpected request to establish sync stream, already connected"); - } - - receivingLines = Connect(establishSyncStream); - break; case FetchCredentials fetchCredentials: if (fetchCredentials.DidExpire) { @@ -629,10 +622,7 @@ async Task HandleInstruction(Instruction instruction) try { await Options.Remote.FetchCredentials(); - controlInvocations?.Emit(new EnqueuedCommand - { - Command = PowerSyncControlCommand.NOTIFY_TOKEN_REFRESHED - }); + notifyTokenRefreshed?.Invoke(); } catch (Exception err) { @@ -641,11 +631,6 @@ async Task HandleInstruction(Instruction instruction) } break; - case CloseSyncStream closeSyncStream: - nestedCts.Cancel(); - hideDisconnectOnRestart = closeSyncStream.HideDisconnect; - logger.LogWarning("Closing stream"); - break; case FlushFileSystem: // ignore break; @@ -657,6 +642,9 @@ async Task HandleInstruction(Instruction instruction) } } + EventStream? controlInvocations = null; + Task? receivingLines = null; + try { var options = new @@ -666,43 +654,131 @@ async Task HandleInstruction(Instruction instruction) include_defaults = resolvedOptions.IncludeDefaultStreams, app_metadata = resolvedOptions.AppMetadata }; - await Control(PowerSyncControlCommand.START, JsonConvert.SerializeObject(options)); + + StreamingSyncRequest? establishRequest = null; + foreach (var startInstruction in await InvokePowerSyncControl(PowerSyncControlCommand.START, JsonConvert.SerializeObject(options))) + { + if (startInstruction is EstablishSyncStream establish) + { + controlInvocations = new EventStream(); + establishRequest = establish.Request; + } + else if (startInstruction is CloseSyncStream) + { + return new StreamingSyncIterationResult { ImmediateRestart = false }; + } + else + { + await HandleInstruction(startInstruction); + } + } + + if (controlInvocations == null) + { + return new StreamingSyncIterationResult { ImmediateRestart = false }; + } + + var invocations = controlInvocations; + + // Subscribe before reading starts, else the (possibly synchronous) "established" + // event is lost. + var commands = invocations.ListenAsync(nestedCts.Token); + receivingLines = ReceiveSyncLines(establishRequest!, invocations, nestedCts.Token); notifyCompletedUploads = () => { - Task.Run(() => + if (!invocations.Closed) { - if (controlInvocations != null && !controlInvocations.Closed) + invocations.Emit(new EnqueuedCommand { - controlInvocations?.Emit(new EnqueuedCommand - { - Command = PowerSyncControlCommand.NOTIFY_CRUD_UPLOAD_COMPLETED - }); - } - }); + Command = PowerSyncControlCommand.NOTIFY_CRUD_UPLOAD_COMPLETED + }); + } }; - handleActiveStreamsChange = () => { - if (controlInvocations != null && !controlInvocations.Closed) + if (!invocations.Closed) { - controlInvocations?.Emit(new EnqueuedCommand + invocations.Emit(new EnqueuedCommand { Command = PowerSyncControlCommand.UPDATE_SUBSCRIPTIONS, Payload = JsonConvert.SerializeObject(activeStreams) }); } }; + notifyTokenRefreshed = () => + { + if (!invocations.Closed) + { + invocations.Emit(new EnqueuedCommand + { + Command = PowerSyncControlCommand.NOTIFY_TOKEN_REFRESHED + }); + } + }; - if (receivingLines != null) + var hadSyncLine = false; + try { - await receivingLines; + await foreach (var command in commands) + { + var close = false; + foreach (var instruction in await InvokePowerSyncControl(command.Command, command.Payload)) + { + if (instruction is EstablishSyncStream) + { + throw new Exception("Received EstablishSyncStream while already connected."); + } + if (instruction is CloseSyncStream closeSyncStream) + { + hideDisconnectOnRestart = closeSyncStream.HideDisconnect; + logger.LogWarning("Closing stream"); + close = true; + break; + } + await HandleInstruction(instruction); + } + + if (!hadSyncLine && + (command.Command == PowerSyncControlCommand.PROCESS_TEXT_LINE || + command.Command == PowerSyncControlCommand.PROCESS_BSON_LINE)) + { + // Triggers a local CRUD upload when the first sync line has been received. + // This allows uploading local changes that have been made while offline or disconnected. + hadSyncLine = true; + TriggerCrudUpload(); + } + + if (close) + { + break; + } + } + } + catch (OperationCanceledException) when (nestedCts.IsCancellationRequested) + { + // Disconnect/abort, stop consuming + } + + if (streamError != null) + { + System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(streamError).Throw(); } } finally { notifyCompletedUploads = null; handleActiveStreamsChange = null; + notifyTokenRefreshed = null; + + nestedCts.Cancel(); + controlInvocations?.Close(); + + if (receivingLines != null) + { + try { await receivingLines; } catch { /* surfaced via streamError */ } + } + await Stop(); } diff --git a/PowerSync/PowerSync.Maui/CHANGELOG.md b/PowerSync/PowerSync.Maui/CHANGELOG.md index 8fb0720..4dd134b 100644 --- a/PowerSync/PowerSync.Maui/CHANGELOG.md +++ b/PowerSync/PowerSync.Maui/CHANGELOG.md @@ -1,5 +1,9 @@ # PowerSync.Maui Changelog +## 0.1.4 (Unreleased) + +- Upstream PowerSync.Common version bump (See Powersync.Common changelog 0.1.4 for more information) + ## 0.1.3 - Upstream PowerSync.Common version bump (See Powersync.Common changelog 0.1.3 for more information) diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationTeardownTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationTeardownTests.cs new file mode 100644 index 0000000..4145d27 --- /dev/null +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationTeardownTests.cs @@ -0,0 +1,126 @@ +namespace PowerSync.Common.Tests.Client.Sync; + +using System.Collections.Concurrent; + +using Newtonsoft.Json; + +using PowerSync.Common.Client; +using PowerSync.Common.Tests.Utils; +using PowerSync.Common.Tests.Utils.Sync; + +/// +/// Verifies how a sync iteration tears down: once an iteration has ended, no +/// further control op is forwarded to the core. The core rejects control ops +/// after its iteration stops (see ), so +/// forwarding a queued op after teardown would surface an unhandled exception. +/// The control loop must therefore stop forwarding as soon as the iteration +/// closes. +/// +/// dotnet test -v n --framework net8.0 --filter "SyncIterationTeardownTests" +/// +public class SyncIterationTeardownTests +{ + /// + /// Establishes the core contract the teardown behaviour relies on: once the + /// iteration has been stopped, forwarding any further control op throws. + /// + [Fact] + public async Task ControlOpAfterStopThrows() + { + var syncService = new MockSyncService(); + var db = syncService.CreateDatabase(); + await db.Init(); + + try + { + Task Control(string op, object? payload = null) => + db.WriteTransaction(async tx => + (await tx.Get("SELECT powersync_control(?, ?) AS r", [op, payload])).r!); + + var startPayload = JsonConvert.SerializeObject(new + { + parameters = new Dictionary(), + active_streams = Array.Empty(), + include_defaults = true, + app_metadata = new Dictionary() + }); + + await Control("start", startPayload); + await Control("stop"); + + var ex = await Assert.ThrowsAnyAsync(() => Control("line_text", "{}")); + Assert.Contains("No iteration is active", ex.Message); + } + finally + { + syncService.Close(); + await db.Close(); + DatabaseUtils.CleanDb(db.Database.Name); + } + } + + /// + /// Disconnecting while sync lines are still queued must end the iteration + /// cleanly: queued control ops are dropped rather than forwarded to a core + /// that no longer has an active iteration. No control exception should escape + /// the iteration task, even across many rapid reconnect cycles. + /// + [Fact(Timeout = 60000)] + public async Task DisconnectWithQueuedLinesEndsCleanly() + { + var unobserved = new ConcurrentBag(); + void Handler(object? sender, UnobservedTaskExceptionEventArgs e) + { + foreach (var ex in e.Exception.Flatten().InnerExceptions) + { + unobserved.Add(ex.Message); + } + e.SetObserved(); + } + + TaskScheduler.UnobservedTaskException += Handler; + try + { + for (int i = 0; i < 12; i++) + { + var syncService = new MockSyncService(); + var db = syncService.CreateDatabase(); + await db.Init(); + + await db.Connect(new TestConnector()); + + // Establish + rapidly enqueue lines so the control queue backs up. + syncService.PushLine(MockDataFactory.Checkpoint(lastOpId: 0, buckets: [])); + for (int j = 0; j < 30; j++) + { + syncService.PushLine(MockDataFactory.Checkpoint(lastOpId: j, buckets: [])); + } + + // Disconnect while lines are still queued (e.g. socket death / reconnect). + await db.Disconnect(); + + syncService.Close(); + await db.Close(); + DatabaseUtils.CleanDb(db.Database.Name); + } + + // Surface any unobserved task exceptions. + for (int k = 0; k < 5; k++) + { + GC.Collect(); + GC.WaitForPendingFinalizers(); + await Task.Delay(50); + } + } + finally + { + TaskScheduler.UnobservedTaskException -= Handler; + } + + var leaked = unobserved.Where(m => m.Contains("No iteration is active")).ToList(); + Assert.True(leaked.Count == 0, + $"Expected no control ops forwarded after teardown, but {leaked.Count} 'No iteration is active' exceptions escaped."); + } + + private record ControlRow(string? r); +} From 431456c323303ed97187a4eaafc9d98a757b7233 Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Tue, 28 Jul 2026 11:53:05 +0200 Subject: [PATCH 2/3] Replaces the `Instruction.IsInterrupting` predicate with a `NonInterruptingInstruction` abstract subclass, matching powersync.dart. --- .../Client/Sync/Stream/CoreInstructions.cs | 28 +++++++------------ .../Stream/StreamingSyncImplementation.cs | 16 ++++++----- 2 files changed, 19 insertions(+), 25 deletions(-) diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs index 1c07c01..e3b5441 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs @@ -11,15 +11,6 @@ namespace PowerSync.Common.Client.Sync.Stream; /// public abstract class Instruction { - /// - /// Whether this instruction starts or stops a sync iteration - /// ( or ). - /// Interrupting instructions are handled by the iteration control loop; all - /// other (non-interrupting) instructions are handled by - /// HandleInstruction. - /// - public virtual bool IsInterrupting => false; - public static Instruction[] ParseInstructions(string rawResponse) { var jsonArray = JArray.Parse(rawResponse); @@ -59,7 +50,12 @@ public static Instruction[] ParseInstructions(string rawResponse) } } -public class LogLine : Instruction +/// +/// An that doesn't start or stop a sync iteration. +/// +public abstract class NonInterruptingInstruction : Instruction { } + +public class LogLine : NonInterruptingInstruction { [JsonProperty("severity")] public string Severity { get; set; } = null!; // "DEBUG", "INFO", "WARNING" @@ -70,13 +66,11 @@ public class LogLine : Instruction public class EstablishSyncStream : Instruction { - public override bool IsInterrupting => true; - [JsonProperty("request")] public StreamingSyncRequest Request { get; set; } = null!; } -public class UpdateSyncStatus : Instruction +public class UpdateSyncStatus : NonInterruptingInstruction { [JsonProperty("status")] public CoreSyncStatus Status { get; set; } = null!; @@ -177,7 +171,7 @@ public class BucketProgress public int TargetCount { get; set; } } -public class FetchCredentials : Instruction +public class FetchCredentials : NonInterruptingInstruction { [JsonProperty("did_expire")] public bool DidExpire { get; set; } @@ -185,14 +179,12 @@ public class FetchCredentials : Instruction public class CloseSyncStream : Instruction { - public override bool IsInterrupting => true; - [JsonProperty("hide_disconnect")] public bool HideDisconnect { get; set; } } -public class FlushFileSystem : Instruction { } -public class DidCompleteSync : Instruction { } +public class FlushFileSystem : NonInterruptingInstruction { } +public class DidCompleteSync : NonInterruptingInstruction { } public class CoreInstructionHelpers { diff --git a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs index 5531f5b..acc06ee 100644 --- a/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs +++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs @@ -573,11 +573,10 @@ async Task Stop() foreach (var instruction in await InvokePowerSyncControl(PowerSyncControlCommand.STOP)) { // Unconditionally ending the iteration, so interrupting instructions don't apply. - if (instruction.IsInterrupting) + if (instruction is NonInterruptingInstruction nonInterrupting) { - continue; + await HandleInstruction(nonInterrupting); } - await HandleInstruction(instruction); } } @@ -588,7 +587,7 @@ async Task InvokePowerSyncControl(string op, object? payload = nu return Instruction.ParseInstructions(rawResponse); } - async Task HandleInstruction(Instruction instruction) + async Task HandleInstruction(NonInterruptingInstruction instruction) { switch (instruction) { @@ -667,9 +666,9 @@ async Task HandleInstruction(Instruction instruction) { return new StreamingSyncIterationResult { ImmediateRestart = false }; } - else + else if (startInstruction is NonInterruptingInstruction nonInterrupting) { - await HandleInstruction(startInstruction); + await HandleInstruction(nonInterrupting); } } @@ -736,7 +735,10 @@ async Task HandleInstruction(Instruction instruction) close = true; break; } - await HandleInstruction(instruction); + if (instruction is NonInterruptingInstruction nonInterrupting) + { + await HandleInstruction(nonInterrupting); + } } if (!hadSyncLine && From ee05b2ea4932474c4576db847ed3f849048d3a73 Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Tue, 28 Jul 2026 12:05:04 +0200 Subject: [PATCH 3/3] Removed core extension test. --- .../Client/Sync/SyncIterationTeardownTests.cs | 43 +------------------ 1 file changed, 2 insertions(+), 41 deletions(-) diff --git a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationTeardownTests.cs b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationTeardownTests.cs index 4145d27..94a3060 100644 --- a/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationTeardownTests.cs +++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationTeardownTests.cs @@ -20,50 +20,11 @@ namespace PowerSync.Common.Tests.Client.Sync; /// public class SyncIterationTeardownTests { - /// - /// Establishes the core contract the teardown behaviour relies on: once the - /// iteration has been stopped, forwarding any further control op throws. - /// - [Fact] - public async Task ControlOpAfterStopThrows() - { - var syncService = new MockSyncService(); - var db = syncService.CreateDatabase(); - await db.Init(); - - try - { - Task Control(string op, object? payload = null) => - db.WriteTransaction(async tx => - (await tx.Get("SELECT powersync_control(?, ?) AS r", [op, payload])).r!); - - var startPayload = JsonConvert.SerializeObject(new - { - parameters = new Dictionary(), - active_streams = Array.Empty(), - include_defaults = true, - app_metadata = new Dictionary() - }); - - await Control("start", startPayload); - await Control("stop"); - - var ex = await Assert.ThrowsAnyAsync(() => Control("line_text", "{}")); - Assert.Contains("No iteration is active", ex.Message); - } - finally - { - syncService.Close(); - await db.Close(); - DatabaseUtils.CleanDb(db.Database.Name); - } - } /// /// Disconnecting while sync lines are still queued must end the iteration - /// cleanly: queued control ops are dropped rather than forwarded to a core - /// that no longer has an active iteration. No control exception should escape - /// the iteration task, even across many rapid reconnect cycles. + /// cleanly, queued control ops are dropped rather than forwarded to a core + /// that no longer has an active iteration. /// [Fact(Timeout = 60000)] public async Task DisconnectWithQueuedLinesEndsCleanly()