diff --git a/src/DodoSSH.Client.Storage/ItemStore.cs b/src/DodoSSH.Client.Storage/ItemStore.cs index ac09465..a08805e 100644 --- a/src/DodoSSH.Client.Storage/ItemStore.cs +++ b/src/DodoSSH.Client.Storage/ItemStore.cs @@ -118,6 +118,65 @@ public sealed class ItemStore( await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); } + /// + /// Writes the server's version of many items at once, in one transaction, in the order given. + /// + /// + /// + /// for a page of pulled changes. A pull used to write each change on its own — + /// a context, a read, a commit — so a first sync of a keychain with a few thousand items was a few + /// thousand SQLite transactions, each one a sync to disk. This is one read for the rows that exist + /// already and one commit for the lot. + /// + /// + /// The same blind overwrite, and the same answer when one item appears more than once: they are applied + /// in order to the same row, so the last one given is the one kept — exactly what writing them one by one + /// would have left. + /// + /// + public async Task SaveManyAsync(IReadOnlyCollection items, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(items); + + if (items.Count == 0) + { + return; + } + + var vaultIds = items.Select(item => item.VaultId).Distinct().ToList(); + var entityIds = items.Select(item => item.EntityId).Distinct().ToList(); + + var context = contexts.CreateDbContext(); + await using var scope = context.ConfigureAwait(false); + + var existing = await context.Set() + .Where(r => vaultIds.Contains(r.VaultId) && entityIds.Contains(r.EntityId)) + .ToListAsync(cancellationToken) + .ConfigureAwait(false); + + var rows = existing.ToDictionary(row => (row.VaultId, row.EntityType, row.EntityId)); + + foreach (var item in items) + { + if (!rows.TryGetValue((item.VaultId, item.EntityType, item.EntityId), out var row)) + { + row = new CachedItemRow + { + VaultId = item.VaultId, + EntityType = item.EntityType, + EntityId = item.EntityId, + }; + + context.Add(row); + rows.Add((item.VaultId, item.EntityType, item.EntityId), row); + } + + Apply(row, item); + } + + await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + } + /// Removes a tombstone whose change has been seen by everything that needed it. /// /// Only ever called for a row that is already a tombstone. Collecting a live item here would make diff --git a/src/DodoSSH.Client.Sync/ItemReconciler.cs b/src/DodoSSH.Client.Sync/ItemReconciler.cs index bddd5e9..6daa8f3 100644 --- a/src/DodoSSH.Client.Sync/ItemReconciler.cs +++ b/src/DodoSSH.Client.Sync/ItemReconciler.cs @@ -546,16 +546,25 @@ internal static class ItemMirror Guid vaultId, SyncChange change, CancellationToken cancellationToken) => - items.SaveAsync( - new StoredItem( - vaultId, - change.EntityType, - change.EntityId, - change.Version, - change.ChangeSequence, - change.Payload, - change.PlaintextFields, - change.Operation == SyncOperation.Delete, - change.UpdatedAt), - cancellationToken); + items.SaveAsync(ToStored(vaultId, change), cancellationToken); + + /// Mirrors a run of changes in one write; see . + internal static Task WriteManyAsync( + ItemStore items, + Guid vaultId, + IReadOnlyList changes, + CancellationToken cancellationToken) => + items.SaveManyAsync(changes.Select(change => ToStored(vaultId, change)).ToList(), cancellationToken); + + private static StoredItem ToStored(Guid vaultId, SyncChange change) => + new( + vaultId, + change.EntityType, + change.EntityId, + change.Version, + change.ChangeSequence, + change.Payload, + change.PlaintextFields, + change.Operation == SyncOperation.Delete, + change.UpdatedAt); } diff --git a/src/DodoSSH.Client.Sync/SyncEngine.cs b/src/DodoSSH.Client.Sync/SyncEngine.cs index 281d9c3..3e162f5 100644 --- a/src/DodoSSH.Client.Sync/SyncEngine.cs +++ b/src/DodoSSH.Client.Sync/SyncEngine.cs @@ -154,9 +154,10 @@ public sealed class SyncEngine continue; } + await ApplyPageAsync(vaultId, response.Changes, report, cancellationToken).ConfigureAwait(false); + foreach (var change in response.Changes) { - await ApplyAsync(vaultId, change, report, cancellationToken).ConfigureAwait(false); report.Pulled++; // Counted apart for the same reason the pushed ones are: a machine reads back the log @@ -310,6 +311,61 @@ public sealed class SyncEngine private static string PulledTypes { get; } = string.Join(',', ItemKinds.SyncedTypes.Select(type => (int)type).Order()); + /// Applies one page of pulled changes, in order, with as few writes as that allows. + /// + /// + /// Each change used to cost a write to mirror it and a read to look for a local edit of the same item — + /// two or more SQLite transactions apiece, which on a first sync of a large keychain was most of the time + /// the sync took. The outbox is read once per page instead, and every run of changes with no local edit + /// waiting is mirrored in a single write. + /// + /// + /// ◆ A change that does have an edit waiting goes through exactly as before, with + /// the run in front of it written first: the reconciler reads the mirror and the outbox as they stand at + /// that change, and both are what the one-by-one loop would have left them as. The one-read snapshot only + /// decides which path a change takes; the path that matters re-reads the outbox for itself. + /// + /// + private async Task ApplyPageAsync( + Guid vaultId, + IReadOnlyList changes, + SyncReportBuilder report, + CancellationToken cancellationToken) + { + if (changes.Count == 0) + { + return; + } + + var waiting = (await outbox.ListAllAsync(vaultId, cancellationToken).ConfigureAwait(false)) + .Select(pending => (pending.EntityType, pending.EntityId)) + .ToHashSet(); + + var run = new List(); + + foreach (var change in changes) + { + if (!reconcilers.ContainsKey(change.EntityType)) + { + // See ApplyAsync: a type this build cannot sync is neither mirrored nor reconciled. + continue; + } + + if (!waiting.Contains((change.EntityType, change.EntityId))) + { + run.Add(change); + continue; + } + + await ItemMirror.WriteManyAsync(items, vaultId, run, cancellationToken).ConfigureAwait(false); + run.Clear(); + + await ApplyAsync(vaultId, change, report, cancellationToken).ConfigureAwait(false); + } + + await ItemMirror.WriteManyAsync(items, vaultId, run, cancellationToken).ConfigureAwait(false); + } + private async Task ApplyAsync( Guid vaultId, SyncChange change, diff --git a/tests/DodoSSH.Client.Storage.Tests/CacheStoreTests.cs b/tests/DodoSSH.Client.Storage.Tests/CacheStoreTests.cs index 9333211..a8bb656 100644 --- a/tests/DodoSSH.Client.Storage.Tests/CacheStoreTests.cs +++ b/tests/DodoSSH.Client.Storage.Tests/CacheStoreTests.cs @@ -128,6 +128,42 @@ public sealed class CacheStoreTests : IAsyncLifetime items.ShouldHaveSingleItem().Version.ShouldBe(2); } + /// + /// A pull's page in one write. What it leaves has to be exactly what the same items saved one at a time + /// would have left: rows that existed replaced, new ones created, and an item given twice ending at the + /// later of the two — a page can carry several changes to the same item. + /// + [Fact] + public async Task SavingMany_LeavesWhatSavingThemOneAtATimeWould() + { + var existing = Guid.CreateVersion7(); + var fresh = Guid.CreateVersion7(); + var twice = Guid.CreateVersion7(); + + await harness.Items.SaveAsync(Item(existing, version: 1, seed: 1), Token); + + await harness.Items.SaveManyAsync( + [ + Item(existing, version: 2, changeSequence: 2, seed: 2), + Item(fresh, version: 1, changeSequence: 3, seed: 3), + Item(twice, version: 1, changeSequence: 4, seed: 4), + Item(twice, version: 2, changeSequence: 5, deleted: true), + ], + Token); + + (await harness.Items.FindAsync(VaultId, SyncEntityType.Host, existing, Token)) + .ShouldNotBeNull().Version.ShouldBe(2); + + (await harness.Items.FindAsync(VaultId, SyncEntityType.Host, fresh, Token)) + .ShouldNotBeNull().Version.ShouldBe(1); + + var last = (await harness.Items.FindAsync(VaultId, SyncEntityType.Host, twice, Token)).ShouldNotBeNull(); + last.Version.ShouldBe(2); + last.IsDeleted.ShouldBeTrue(); + + (await harness.Items.ListAsync(VaultId, SyncEntityType.Host, true, Token)).Count.ShouldBe(3); + } + [Fact] public async Task ATombstone_IsHiddenFromTheListButStillFindable() {