From b761db536e90356e382e5083a37f6c4a443dcb2a Mon Sep 17 00:00:00 2001 From: Austin Hale Date: Wed, 10 Jun 2026 01:18:43 -0700 Subject: [PATCH 1/3] fix: persist attachment state per record and don't hold the context across transfers --- .../Attachments/SyncingService.cs | 31 ++++++++----------- 1 file changed, 13 insertions(+), 18 deletions(-) diff --git a/PowerSync/PowerSync.Common/Attachments/SyncingService.cs b/PowerSync/PowerSync.Common/Attachments/SyncingService.cs index 12dcaea..7dfdf74 100644 --- a/PowerSync/PowerSync.Common/Attachments/SyncingService.cs +++ b/PowerSync/PowerSync.Common/Attachments/SyncingService.cs @@ -113,40 +113,36 @@ private async Task StopSyncInternalAsync() /// Runs one sync pass: fetches active attachments, processes them, then prunes archived rows. /// /// A task that completes when the pass has finished. - public Task RunSyncPassAsync() => attachmentService.WithContextAsync(async ctx => + public async Task RunSyncPassAsync() { - var active = await ctx.GetActiveAttachmentsAsync(); - await ProcessAttachmentsAsync(active, ctx); - await DeleteArchivedAttachmentsAsync(ctx); - }); + var active = await attachmentService.WithContextAsync(ctx => ctx.GetActiveAttachmentsAsync()); + await ProcessAttachmentsAsync(active); + await attachmentService.WithContextAsync(ctx => DeleteArchivedAttachmentsAsync(ctx)); + } /// - /// Processes attachments based on their state. Updates are saved in a single batch. + /// Processes attachments based on their state. Each state change is persisted as soon as its + /// transfer completes. /// /// Attachment records to process. - /// Attachment context for database operations. - /// A task that completes once all attachments have been processed and saved. - public async Task ProcessAttachmentsAsync(IReadOnlyList attachments, AttachmentContext context) + /// A task that completes once all attachments have been processed. + public async Task ProcessAttachmentsAsync(IReadOnlyList attachments) { - var updates = new List(); - foreach (var attachment in attachments) { Attachment? changed = attachment.State switch { AttachmentState.QueuedUpload => await UploadAttachmentAsync(attachment), AttachmentState.QueuedDownload => await DownloadAttachmentAsync(attachment), - AttachmentState.QueuedDelete => await DeleteAttachmentAsync(attachment, context), + AttachmentState.QueuedDelete => await DeleteAttachmentAsync(attachment), _ => null, }; if (changed is not null) { - updates.Add(changed); + await attachmentService.WithContextAsync(ctx => ctx.SaveAttachmentsAsync([changed])); } } - - await context.SaveAttachmentsAsync(updates); } /// @@ -231,9 +227,8 @@ public async Task ProcessAttachmentsAsync(IReadOnlyList attachments, /// On failure, defers to or archives. /// /// The attachment to delete. - /// The attachment context for database operations. /// The archived attachment, or null on success or retry. - public async Task DeleteAttachmentAsync(Attachment attachment, AttachmentContext context) + public async Task DeleteAttachmentAsync(Attachment attachment) { try { @@ -243,7 +238,7 @@ public async Task ProcessAttachmentsAsync(IReadOnlyList attachments, await localStorage.DeleteFileAsync(attachment.LocalUri); } - await context.DeleteAttachmentAsync(attachment.Id); + await attachmentService.WithContextAsync(ctx => ctx.DeleteAttachmentAsync(attachment.Id)); return null; } catch (Exception error) From ac9cd433056a16570389127af0de3b32608fb32a Mon Sep 17 00:00:00 2001 From: Austin Hale Date: Wed, 10 Jun 2026 02:22:34 -0700 Subject: [PATCH 2/3] fix: guard attachment sync against concurrent queue operations --- .../Attachments/AttachmentQueue.cs | 3 +- .../Attachments/SyncingService.cs | 53 +++++++++++++++---- 2 files changed, 44 insertions(+), 12 deletions(-) diff --git a/PowerSync/PowerSync.Common/Attachments/AttachmentQueue.cs b/PowerSync/PowerSync.Common/Attachments/AttachmentQueue.cs index fa55bc8..081ee37 100644 --- a/PowerSync/PowerSync.Common/Attachments/AttachmentQueue.cs +++ b/PowerSync/PowerSync.Common/Attachments/AttachmentQueue.cs @@ -122,8 +122,9 @@ public async Task StartSyncAsync() /// called manually to await an immediate sync pass (e.g. before shutting down). For /// "fire and forget", use . /// + /// The cancellation token observed between records. /// A task that completes when the sync pass has finished. - public Task SyncStorageAsync() => _syncingService.RunSyncPassAsync(); + public Task SyncStorageAsync(CancellationToken ct = default) => _syncingService.RunSyncPassAsync(ct); /// /// Stops the attachment synchronization process. Cancels the sync pipeline, stops the periodic diff --git a/PowerSync/PowerSync.Common/Attachments/SyncingService.cs b/PowerSync/PowerSync.Common/Attachments/SyncingService.cs index 7dfdf74..c3b125c 100644 --- a/PowerSync/PowerSync.Common/Attachments/SyncingService.cs +++ b/PowerSync/PowerSync.Common/Attachments/SyncingService.cs @@ -22,6 +22,7 @@ internal sealed class SyncingService( ILogger logger) { private readonly SemaphoreSlim _startStopLock = new(1, 1); + private readonly SemaphoreSlim _syncPassLock = new(1, 1); private CancellationTokenSource? _internalCts; private Channel? _syncSignals; @@ -112,12 +113,21 @@ private async Task StopSyncInternalAsync() /// /// Runs one sync pass: fetches active attachments, processes them, then prunes archived rows. /// + /// The cancellation token observed between records. /// A task that completes when the pass has finished. - public async Task RunSyncPassAsync() + public async Task RunSyncPassAsync(CancellationToken ct = default) { - var active = await attachmentService.WithContextAsync(ctx => ctx.GetActiveAttachmentsAsync()); - await ProcessAttachmentsAsync(active); - await attachmentService.WithContextAsync(ctx => DeleteArchivedAttachmentsAsync(ctx)); + await _syncPassLock.WaitAsync(ct); + try + { + var active = await attachmentService.WithContextAsync(ctx => ctx.GetActiveAttachmentsAsync()); + await ProcessAttachmentsAsync(active, ct); + await attachmentService.WithContextAsync(ctx => DeleteArchivedAttachmentsAsync(ctx)); + } + finally + { + _syncPassLock.Release(); + } } /// @@ -125,11 +135,14 @@ public async Task RunSyncPassAsync() /// transfer completes. /// /// Attachment records to process. + /// The cancellation token observed between records. /// A task that completes once all attachments have been processed. - public async Task ProcessAttachmentsAsync(IReadOnlyList attachments) + public async Task ProcessAttachmentsAsync(IReadOnlyList attachments, CancellationToken ct = default) { foreach (var attachment in attachments) { + ct.ThrowIfCancellationRequested(); + Attachment? changed = attachment.State switch { AttachmentState.QueuedUpload => await UploadAttachmentAsync(attachment), @@ -140,7 +153,14 @@ public async Task ProcessAttachmentsAsync(IReadOnlyList attachments) if (changed is not null) { - await attachmentService.WithContextAsync(ctx => ctx.SaveAttachmentsAsync([changed])); + await attachmentService.WithContextAsync(async ctx => + { + var current = await ctx.GetAttachmentAsync(attachment.Id); + if (current?.State == attachment.State) + { + await ctx.SaveAttachmentsAsync([changed]); + } + }); } } } @@ -233,12 +253,23 @@ public async Task ProcessAttachmentsAsync(IReadOnlyList attachments) try { await remoteStorage.DeleteFileAsync(attachment); - if (attachment.LocalUri is not null) + + await attachmentService.WithContextAsync(async ctx => { - await localStorage.DeleteFileAsync(attachment.LocalUri); - } + var current = await ctx.GetAttachmentAsync(attachment.Id); + if (current?.State != AttachmentState.QueuedDelete) + { + return; + } + + if (attachment.LocalUri is not null) + { + await localStorage.DeleteFileAsync(attachment.LocalUri); + } + + await ctx.DeleteAttachmentAsync(attachment.Id); + }); - await attachmentService.WithContextAsync(ctx => ctx.DeleteAttachmentAsync(attachment.Id)); return null; } catch (Exception error) @@ -321,7 +352,7 @@ private async Task SyncSignalConsumerAsync(CancellationToken ct) try { - await RunSyncPassAsync(); + await RunSyncPassAsync(ct); } catch (OperationCanceledException) { From 2f87dd19901c62df30d553bd94dc0a40a34267a7 Mon Sep 17 00:00:00 2001 From: Austin Hale Date: Wed, 10 Jun 2026 02:55:01 -0700 Subject: [PATCH 3/3] fix: tighten attachment state handling and file deletion --- .../PowerSync.Common/Attachments/SyncingService.cs | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/PowerSync/PowerSync.Common/Attachments/SyncingService.cs b/PowerSync/PowerSync.Common/Attachments/SyncingService.cs index c3b125c..8e93865 100644 --- a/PowerSync/PowerSync.Common/Attachments/SyncingService.cs +++ b/PowerSync/PowerSync.Common/Attachments/SyncingService.cs @@ -143,7 +143,8 @@ public async Task ProcessAttachmentsAsync(IReadOnlyList attachments, { ct.ThrowIfCancellationRequested(); - Attachment? changed = attachment.State switch + var snapshotState = attachment.State; + Attachment? changed = snapshotState switch { AttachmentState.QueuedUpload => await UploadAttachmentAsync(attachment), AttachmentState.QueuedDownload => await DownloadAttachmentAsync(attachment), @@ -156,7 +157,7 @@ public async Task ProcessAttachmentsAsync(IReadOnlyList attachments, await attachmentService.WithContextAsync(async ctx => { var current = await ctx.GetAttachmentAsync(attachment.Id); - if (current?.State == attachment.State) + if (current?.State == snapshotState) { await ctx.SaveAttachmentsAsync([changed]); } @@ -253,6 +254,10 @@ await attachmentService.WithContextAsync(async ctx => try { await remoteStorage.DeleteFileAsync(attachment); + if (attachment.LocalUri is not null) + { + await localStorage.DeleteFileAsync(attachment.LocalUri); + } await attachmentService.WithContextAsync(async ctx => { @@ -262,11 +267,6 @@ await attachmentService.WithContextAsync(async ctx => return; } - if (attachment.LocalUri is not null) - { - await localStorage.DeleteFileAsync(attachment.LocalUri); - } - await ctx.DeleteAttachmentAsync(attachment.Id); });