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..e3b5441 100644
--- a/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs
+++ b/PowerSync/PowerSync.Common/Client/Sync/Stream/CoreInstructions.cs
@@ -11,7 +11,6 @@ namespace PowerSync.Common.Client.Sync.Stream;
///
public abstract class Instruction
{
-
public static Instruction[] ParseInstructions(string rawResponse)
{
var jsonArray = JArray.Parse(rawResponse);
@@ -51,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"
@@ -66,7 +70,7 @@ public class EstablishSyncStream : Instruction
public StreamingSyncRequest Request { get; set; } = null!;
}
-public class UpdateSyncStatus : Instruction
+public class UpdateSyncStatus : NonInterruptingInstruction
{
[JsonProperty("status")]
public CoreSyncStatus Status { get; set; } = null!;
@@ -167,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; }
@@ -179,8 +183,8 @@ public class CloseSyncStream : Instruction
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 0dc6417..acc06ee 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,42 +539,55 @@ 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 is NonInterruptingInstruction nonInterrupting)
+ {
+ await HandleInstruction(nonInterrupting);
+ }
+ }
}
- 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));
+ return Instruction.ParseInstructions(rawResponse);
}
- async void HandleInstructions(Instruction[] instructions)
- {
- foreach (var instruction in instructions)
- {
- await HandleInstruction(instruction);
- }
- }
-
- async Task HandleInstruction(Instruction instruction)
+ async Task HandleInstruction(NonInterruptingInstruction instruction)
{
switch (instruction)
{
@@ -608,14 +608,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 +621,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 +630,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 +641,9 @@ async Task HandleInstruction(Instruction instruction)
}
}
+ EventStream? controlInvocations = null;
+ Task? receivingLines = null;
+
try
{
var options = new
@@ -666,43 +653,134 @@ 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 if (startInstruction is NonInterruptingInstruction nonInterrupting)
+ {
+ await HandleInstruction(nonInterrupting);
+ }
+ }
+
+ 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;
+ }
+ if (instruction is NonInterruptingInstruction nonInterrupting)
+ {
+ await HandleInstruction(nonInterrupting);
+ }
+ }
+
+ 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..94a3060
--- /dev/null
+++ b/Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationTeardownTests.cs
@@ -0,0 +1,87 @@
+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
+{
+
+ ///
+ /// 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.
+ ///
+ [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);
+}